agent-skills-context-engineering

name	agent-skills-context-engineering
description	Master context engineering principles for building production-grade AI agent systems with effective context management, multi-agent architectures, and memory systems.
triggers	["build agent system with context engineering","optimize agent context window usage","implement multi-agent architecture","design agent memory system","compress agent context for long sessions","debug agent context degradation","evaluate agent performance with LLM-as-judge","build hosted coding agent with sandboxes"]

Agent Skills for Context Engineering

Skill by ara.so — AI Agent Skills collection.

A comprehensive collection of Agent Skills for context engineering, multi-agent architectures, and production agent systems. This skill teaches principles for managing LLM context windows, designing effective agent architectures, and building production-grade agent systems.

What This Project Does

Agent Skills for Context Engineering provides battle-tested patterns for:

• Context Management: Managing limited attention budgets, avoiding lost-in-middle degradation
• Multi-Agent Systems: Orchestrator, peer-to-peer, and hierarchical architectures
• Memory Systems: Short-term, long-term, and graph-based memory patterns
• Tool Design: Building tools that agents can use effectively
• Evaluation: LLM-as-judge frameworks for measuring agent quality
• Production Systems: Hosted agents with sandboxed VMs and multiplayer support

Unlike prompt engineering (crafting instructions), context engineering addresses holistic curation of all information in the context window: system prompts, tool definitions, retrieved documents, message history, and tool outputs.

Installation

For Claude Code (Recommended)

Step 1: Add the Marketplace

/plugin marketplace add muratcankoylan/Agent-Skills-for-Context-Engineering

Step 2: Install the Plugin

/plugin install context-engineering@context-engineering-marketplace

Or browse and install:

1. Select Browse and install plugins
2. Select context-engineering-marketplace
3. Select context-engineering
4. Select Install now

For Cursor (Open Plugins)

Add to your .cursor/plugins.json:

{
  "plugins": [
    {
      "name": "context-engineering",
      "repository": "muratcankoylan/Agent-Skills-for-Context-Engineering"
    }
  ]
}

For Individual Skills

Copy specific skills to your project:

# Create skills directory
mkdir -p .claude/skills

# Add a specific skill (example: context-fundamentals)
curl -o .claude/skills/context-fundamentals.md \
  https://raw.githubusercontent.com/muratcankoylan/Agent-Skills-for-Context-Engineering/main/skills/context-fundamentals/SKILL.md

Available skills: context-fundamentals, context-degradation, context-compression, context-optimization, latent-briefing, multi-agent-patterns, memory-systems, tool-design, filesystem-context, hosted-agents, evaluation, advanced-evaluation, project-development, bdi-mental-states

Core Concepts

Context Window Management

The fundamental challenge: context windows are constrained by attention mechanics, not raw token capacity.

Key degradation patterns:

• Lost-in-the-middle: Models lose track of information in the middle of long contexts
• U-shaped attention: Strong attention at beginning and end, weak in middle
• Attention scarcity: As context grows, attention per token decreases

Solution: Find the smallest possible set of high-signal tokens.

Progressive Disclosure

Load information only when needed:

# skills/__init__.py - Lazy loading pattern
class SkillRegistry:
    def __init__(self):
        self._skills = {}
        self._loaded = set()
    
    def get_skill_summary(self, skill_name: str) -> dict:
        """Load only name and description initially."""
        return {
            "name": skill_name,
            "description": self._get_description(skill_name)
        }
    
    def load_skill(self, skill_name: str) -> dict:
        """Load full skill content only when activated."""
        if skill_name not in self._loaded:
            self._skills[skill_name] = self._read_skill_file(skill_name)
            self._loaded.add(skill_name)
        return self._skills[skill_name]

Context Compression Strategies

Sliding Window:

def sliding_window_context(messages: list, window_size: int = 10) -> list:
    """Keep only recent messages."""
    if len(messages) <= window_size:
        return messages
    
    # Always keep system message
    system_msgs = [m for m in messages if m["role"] == "system"]
    recent_msgs = messages[-window_size:]
    
    return system_msgs + recent_msgs

Summarization:

async def compress_with_summary(messages: list, llm_client) -> list:
    """Compress old messages into summary."""
    if len(messages) < 20:
        return messages
    
    # Keep recent messages uncompressed
    to_compress = messages[1:-10]  # Skip system message and recent 10
    recent = messages[-10:]
    
    # Generate summary
    summary_prompt = f"Summarize these messages concisely:\n{to_compress}"
    summary = await llm_client.complete(summary_prompt)
    
    return [
        messages[0],  # System message
        {"role": "assistant", "content": f"[Summary of previous conversation: {summary}]"},
        *recent
    ]

Multi-Agent Patterns

Orchestrator Pattern

Single coordinator delegates to specialized workers:

from typing import List, Dict

class OrchestratorAgent:
    def __init__(self, workers: Dict[str, Agent]):
        self.workers = workers
    
    async def process_task(self, task: str) -> str:
        # Determine which worker to use
        worker_name = await self._route_task(task)
        worker = self.workers[worker_name]
        
        # Delegate to worker with minimal context
        result = await worker.execute(task)
        
        return result
    
    async def _route_task(self, task: str) -> str:
        """Use LLM to determine which worker handles task."""
        routing_prompt = f"""Given this task: {task}

Available workers:
- code_writer: Writes and modifies code
- researcher: Gathers information and analyzes data
- reviewer: Reviews code and provides feedback

Which worker should handle this? Respond with just the worker name."""
        
        return await self.llm.complete(routing_prompt)

# Usage
orchestrator = OrchestratorAgent({
    "code_writer": CodeWriterAgent(),
    "researcher": ResearcherAgent(),
    "reviewer": ReviewerAgent()
})

result = await orchestrator.process_task()

Peer-to-Peer Pattern

Agents collaborate directly:

class PeerAgent:
    def __init__(self, name: str, peers: List['PeerAgent']):
        self.name = name
        self.peers = peers
        self.messages = []
    
    async def broadcast(self, message: str):
        """Send message to all peers."""
        for peer in self.peers:
            await peer.receive(self.name, message)
    
    async def receive(self, sender: str, message: str):
        """Receive message from peer."""
        self.messages.append({
            "from": sender,
            "content": message,
            "timestamp": time.time()
        })

Memory Systems

Short-term Memory (Working Context)

class WorkingMemory:
    def __init__(self, max_items: int = 5):
        self.items = []
        self.max_items = max_items
    
    def add(self, item: dict):
        """Add item, removing oldest if at capacity."""
        self.items.append(item)
        if len(self.items) > self.max_items:
            self.items.pop(0)
    
    def get_context(self) -> str:
        """Format for inclusion in prompt."""
        return "\n".join([
            f"- {item['key']}: {item['value']}"
            for item in self.items
        ])

Long-term Memory (Retrieval)

import chromadb
from typing import List, Dict

class LongTermMemory:
    def __init__(self):
        self.client = chromadb.Client()
        self.collection = self.client.create_collection("agent_memory")
    
    def store(self, content: str, metadata: dict = None):
        """Store information for later retrieval."""
        self.collection.add(
            documents=[content],
            metadatas=[metadata or {}],
            ids=[str(hash(content))]
        )
    
    def recall(self, query: str, n_results: int = 3) -> List[Dict]:
        """Retrieve relevant memories."""
        results = self.collection.query(
            query_texts=[query],
            n_results=n_results
        )
        
        return [
            {
                "content": doc,
                "metadata": meta
            }
            for doc, meta in zip(results['documents'][0], results['metadatas'][0])
        ]

Graph-based Memory

import networkx as nx

class GraphMemory:
    def __init__(self):
        self.graph = nx.DiGraph()
    
    def add_entity(self, entity: str, properties: dict):
        """Add or update entity node."""
        self.graph.add_node(entity, **properties)
    
    def add_relation(self, from_entity: str, to_entity: str, relation: str):
        """Add relationship between entities."""
        self.graph.add_edge(from_entity, to_entity, relation=relation)
    
    def get_neighbors(self, entity: str, max_depth: int = 2) -> dict:
        """Get connected entities within depth."""
        if entity not in self.graph:
            return {}
        
        # BFS to find neighbors
        neighbors = {}
        for node in nx.single_source_shortest_path_length(
            self.graph, entity, cutoff=max_depth
        ):
            neighbors[node] = self.graph.nodes[node]
        
        return neighbors

Tool Design Principles

Minimal Interface

from typing import Dict, Any

def search_documentation(query: str, max_results: int = 5) -> list[Dict[str, Any]]:
    """Search documentation with minimal parameters.
    
    Args:
        query: Search query string
        max_results: Maximum number of results to return (default: 5)
    
    Returns:
        List of matching documentation sections with title and content
    
    Example:
        results = search_documentation("authentication")
    """
    # Implementation
    pass

Clear Output Format

def analyze_code(code: str) -> dict:
    """Analyze code and return structured results.
    
    Returns:
        {
            "issues": [{"line": int, "severity": str, "message": str}],
            "metrics": {"complexity": int, "lines": int},
            "suggestions": [str]
        }
    """
    return {
        "issues": [
            {"line": 15, "severity": "warning", "message": "Unused variable 'x'"}
        ],
        "metrics": {
            "complexity": 7,
            "lines": 42
        },
        "suggestions": [
            "Consider extracting this logic into a separate function"
        ]
    }

Context Offloading

import json
from pathlib import Path

def analyze_large_dataset(data_path: str, output_dir: str = ".agent_context") -> str:
    """Analyze data and write detailed results to file.
    
    Returns reference to results file instead of full data in context.
    """
    # Create context directory
    Path(output_dir).mkdir(exist_ok=True)
    
    # Analyze data
    results = perform_analysis(data_path)
    
    # Write detailed results to file
    results_file = f"{output_dir}/analysis_results.json"
    with open(results_file, 'w') as f:
        json.dump(results, f, indent=2)
    
    # Return only summary in context
    summary = {
        "total_records": results["count"],
        "key_findings": results["top_insights"][:3],
        "full_results": results_file
    }
    
    return f"Analysis complete. Summary: {summary}\nFull results in {results_file}"

Filesystem-based Context Management

Dynamic Discovery

from pathlib import Path
import yaml

def discover_tools(tools_dir: str = ".agent_tools") -> dict:
    """Dynamically discover available tools from filesystem."""
    tools = {}
    
    for tool_file in Path(tools_dir).glob("*.yaml"):
        with open(tool_file) as f:
            tool_spec = yaml.safe_load(f)
            tools[tool_spec["name"]] = tool_spec
    
    return tools

# Tool definition file: .agent_tools/github_search.yaml
"""
name: github_search
description: Search GitHub repositories
parameters:
  - name: query
    type: string
    required: true
  - name: language
    type: string
    required: false
"""

Plan Persistence

import json
from datetime import datetime

class PlanTracker:
    def __init__(self, plan_file: str = ".agent_context/current_plan.json"):
        self.plan_file = plan_file
    
    def save_plan(self, steps: list):
        """Persist plan to disk."""
        plan = {
            "created_at": datetime.now().isoformat(),
            "steps": steps,
            "completed": []
        }
        
        with open(self.plan_file, 'w') as f:
            json.dump(plan, f, indent=2)
    
    def mark_complete(self, step_index: int):
        """Mark step as complete."""
        with open(self.plan_file, 'r') as f:
            plan = json.load(f)
        
        plan["completed"].append(step_index)
        
        with open(self.plan_file, 'w') as f:
            json.dump(plan, f, indent=2)
    
    def get_next_step(self) -> dict:
        """Get next incomplete step."""
         (.plan_file, )  f:
            plan = json.load(f)
        
         i, step  (plan[]):
             i   plan[]:
                 {: i, : step}
        
         

LLM-as-Judge Evaluation

Direct Scoring

from typing import List, Dict

async def score_response(
    response: str,
    criteria: List[Dict[str, any]],
    llm_client
) -> dict:
    """Score response against weighted criteria.
    
    Args:
        response: Response to evaluate
        criteria: List of {name, description, weight, scale}
        llm_client: LLM client for evaluation
    
    Returns:
        {
            "total_score": float,
            "criteria_scores": [{name, score, reasoning}]
        }
    """
    scores = []
    
    for criterion in criteria:
        prompt = f"""Evaluate this response on {criterion['name']}:

{criterion['description']}

Response to evaluate:
{response}

Score from 1-{criterion['scale']} and explain your reasoning.
Format: SCORE: X | REASONING: explanation"""
        
        evaluation = await llm_client.complete(prompt)
        
        # Parse score and reasoning
        score_line = [l for l in evaluation.split('\n') if 'SCORE:' in l][0]
        score = int(score_line.split('SCORE:')[1].split('|')[0].strip())
        reasoning = evaluation.split('REASONING:')[1].strip()
        
        scores.append({
            "name": criterion['name'],
            : score,
            : criterion[],
            : criterion[],
            : reasoning
        })
    
    
    total = (s[] / s[] * s[]  s  scores)
    total_weight = (c[]  c  criteria)
    
     {
        : total / total_weight,
        : scores
    }


result =  score_response(
    response=,
    criteria=[
        {
            : ,
            : ,
            : ,
            : 
        },
        {
            : ,
            : ,
            : ,
            : 
        }
    ],
    llm_client=client
)

Pairwise Comparison

async def compare_responses(
    response_a: str,
    response_b: str,
    criteria: str,
    llm_client,
    mitigate_bias: bool = True
) -> dict:
    """Compare two responses with position bias mitigation.
    
    Args:
        response_a: First response
        response_b: Second response
        criteria: Evaluation criteria
        llm_client: LLM client
        mitigate_bias: Run comparison both ways and aggregate
    
    Returns:
        {"winner": "A" | "B" | "tie", "reasoning": str, "confidence": float}
    """
    async def single_comparison(first: str, second: str) -> dict:
        prompt = f"""Compare these two responses based on: {criteria}

Response 1:
{first}

Response 2:
{second}

Which is better? Respond with: WINNER: [1|2|TIE] | REASONING: explanation"""
        
        result = await llm_client.complete(prompt)
        winner_line = [l for l in result.split('\n') if 'WINNER:' in l][0]
        winner = winner_line.split('WINNER:')[1].split('|')[0].strip()
        reasoning = result.split('REASONING:')[1].strip()
        
        return {"winner": winner, "reasoning": reasoning}
    
    # First comparison (A then B)
    comp1 = await single_comparison(response_a, response_b)
    
      mitigate_bias:
         {
            :   comp1[] ==     comp1[] ==   ,
            : comp1[],
            : 
        }
    
    
    comp2 =  single_comparison(response_b, response_a)
    
    
     comp1[] ==   comp2[] == :
         {: , : comp1[], : }
     comp1[] ==   comp2[] == :
         {: , : comp1[], : }
     comp1[] ==   comp2[] == :
         {: , : , : }
    :
         {: , : , : }

Hosted Agents (Sandboxed Execution)

Modal-based Background Agent

import modal

# Create Modal app
app = modal.App("coding-agent")

# Define image with dependencies
image = modal.Image.debian_slim().pip_install(
    "anthropic",
    "requests"
)

@app.function(
    image=image,
    secrets=[modal.Secret.from_name("anthropic-api-key")],
    timeout=3600
)
async def run_coding_task(task: str, files: dict) -> dict:
    """Execute coding task in sandboxed environment.
    
    Args:
        task: Task description
        files: Initial files as {path: content}
    
    Returns:
        {
            "status": "success" | "error",
            "files": {path: content},
            "output": str
        }
    """
    import os
    from anthropic import Anthropic
    
    # Initialize Claude
    client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
    
    # Write initial files
    for path, content in files.items():
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, 'w') as f:
            f.write(content)
    
    # Execute task with agent
    messages = [
        {
            "role": "user",
            "content": f"Complete this task: {task}\n\nAvailable files: {list(files.keys())}"
        }
    ]
    
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=,
        messages=messages
    )
    
    
    output_files = {}
     path  files.keys():
         os.path.exists(path):
             (path, )  f:
                output_files[path] = f.read()
    
     {
        : ,
        : output_files,
        : response.content[].text
    }



 ():
    result = run_coding_task.remote(
        task=,
        files={
            : 
        }
    )
    
    (result[])
    ()
     path, content  result[].items():
        ()
        (content)

Common Patterns

Task-Model Fit Analysis

def analyze_task_model_fit(task_description: str) -> dict:
    """Determine if LLM is appropriate for task.
    
    Returns:
        {
            "recommended": bool,
            "model_size": "small" | "medium" | "large",
            "approach": "single_shot" | "chain_of_thought" | "multi_agent",
            "reasoning": str
        }
    """
    # Decision tree
    is_structured = "json" in task_description.lower() or "schema" in task_description.lower()
    is_creative = any(w in task_description.lower() for w in ["generate", "write", "create"])
    is_complex = any(w in task_description.lower() for w in ["analyze", "reason", "solve"])
    
    if is_structured:
        return {
            "recommended": True,
            "model_size": "small",
            "approach": "single_shot",
            "reasoning": "Structured output tasks work well with smaller models"
        }
    elif is_complex:
        return {
            "recommended": True,
            "model_size": "large",
            "approach": "chain_of_thought",
            : 
        }
     is_creative:
         {
            : ,
            : ,
            : ,
            : 
        }
    :
         {
            : ,
            : ,
            : ,
            : 
        }

Append-Only Memory Pattern

import json
from datetime import datetime
from pathlib import Path

class AppendOnlyMemory:
    """JSONL-based memory with schema-first line for agent parsing."""
    
    def __init__(self, file_path: str, schema: dict):
        self.file_path = Path(file_path)
        self.schema = schema
        
        # Create file with schema if doesn't exist
        if not self.file_path.exists():
            self.file_path.parent.mkdir(parents=True, exist_ok=True)
            with open(self.file_path, 'w') as f:
                f.write(json.dumps({"_schema": schema}) + '\n')
    
    def append(self, data: dict):
        """Append entry with timestamp."""
        entry = {
            **data,
            "_timestamp": datetime.now().isoformat()
        }
        
        with open(self.file_path, 'a') as f:
            f.write(json.dumps(entry) + '\n')
    
    def read_recent(self, n: int = 10) -> :
        
         (.file_path, )  f:
            lines = f.readlines()
        
        
        entries = [json.loads(line)  line  lines[:]]
         entries[-n:]
    
     () -> :
        
         (.file_path, )  f:
            lines = f.readlines()
        
        entries = [json.loads(line)  line  lines[:]]
         [e  e  entries  filter_fn(e)]


memory = AppendOnlyMemory(
    ,
    schema={
        : ,
        : ,
        : 
    }
)

memory.append({
    : ,
    : ,
    : 
})

recent = memory.read_recent()

Troubleshooting

Lost-in-Middle Degradation

Symptom: Agent ignores information from middle of context.

Solution: Use U-shaped context organization:

def organize_context(system_msg: str, key_facts: list, messages: list) -> list:
    """Place critical info at beginning and end."""
    return [
        {"role": "system", "content": system_msg},
        # Key facts at the top
        {"role": "system", "content": "Key facts:\n" + "\n".join(key_facts)},
        # Conversation in middle
        *messages[:-3],
        # Recent context at the end
        *messages[-3:],
        # Reminder of key facts at very end
        {"role": "system", "content": "Remember:\n" + "\n".join(key_facts[:3])}
    ]

Context Clash

Symptom: Agent confused by contradictory information.

Solution: Timestamp and prioritize information:

def merge_contexts(contexts: list[dict]) -> str:
    """Merge contexts with priority and timestamps."""
    # Sort by priority (higher first) then timestamp (newer first)
    sorted_contexts = sorted(
        contexts,
        key=lambda x: (-x.get('priority', 0), -x.get('timestamp', 0))
    )
    
    merged = []
    for ctx in sorted_contexts:
        timestamp_str = datetime.fromtimestamp(ctx['timestamp']).strftime('%Y-%m-%d %H:%M')
        merged.append(f"[{timestamp_str}, priority={ctx.get('priority', 0)}] {ctx['content']}")
    
    return "\n\n".join(merged)

Attention Scarcity

Symptom: Performance degrades with long contexts.

Solution: Implement KV-cache compaction:

def compact_messages(messages: list, target_tokens: int) -> list:
    """Compress messages to target token budget."""
    current_tokens = sum(estimate_tokens(m["content"]) for m in messages)
    
    if current_tokens <= target_tokens:
        return messages
    
    # Keep system message and recent messages
    system = [m for m in messages if m["role"] == "system"]
    recent = messages[-5:]
    
    # Compress middle messages
    middle = messages[len(system):-5]
    compression_ratio = (target_tokens - estimate_tokens(system + recent)) / estimate_tokens(middle)
    
    if compression_ratio < 0.5:
        # Aggressive summarization needed
        summary = summarize_messages(middle)
        return system + [{"role": "assistant", "content": f"[Earlier: {summary}]"}] + recent
    else:
        # Keep important messages, drop others
        important = extract_important_messages(middle, keep_ratio=compression_ratio)
        return system + important + recent

def estimate_tokens(text: str) -> int:
    """Rough estimate: 1 token per 4 characters."""
    return len(text) // 

Tool Output Bloat

Symptom: Tool outputs fill context window.

Solution: Offload to filesystem:

def execute_tool_with_offloading(
    tool_name: str,
    params: dict,
    context_dir: str = ".agent_context/tool_outputs"
) -> str:
    """Execute tool and offload detailed results."""
    Path(context_dir).mkdir(parents=True, exist_ok=True)
    
    # Execute tool
    result = execute_tool(tool_name, params)
    
    # Write full result to file
    output_file = f"{context_dir}/{tool_name}_{int(time.time())}.json"
    with open(output_file, 'w') as f:
        json.dump(result, f, indent=2)
    
    # Return summary + file reference
    summary = {
        "status": result.get("status"),
        "key_findings": result.get("summary", "See file for details"),
        "record_count": len(result.get("data", [])),
        "full_output": output_file
    }
    
    return json.dumps(summary)

Best Practices

1. Progressive Disclosure: Load information only when needed
2. Context Budget: Track token usage, compress proactively
3. Tool Minimalism: Design tools with minimal interface, maximum clarity
4. Filesystem Offloading: Keep detailed data out of context
5. Evaluation First: Build evaluation before building features
6. Platform Agnostic: Focus on transferable principles over vendor APIs
7. Append-Only Memory: Use JSONL with schema-first for agent-friendly persistence
8. U-shaped Organization: Place critical info at start and end of context

Resources

• Repository
• DeepWiki Documentation
• Cursor Plugin Directory
• Examples Directory

License

MIT