Context Engineering for AI Agents: The 2026 Practitioner's Deep Dive

AI Agent = LLM Model + Harness

import anthropic
from typing import Generator

client = anthropic.Anthropic()

def run_agent_loop(
    task: str,
    tools: list[dict],
    system_prompt: str,
    context_policy: "ContextPolicy",
    max_iterations: int = 50,
) -> str:
    """
    Minimal production harness with context policy enforcement.

    The context_policy object handles:
    - Which message history to include
    - When to trigger compaction
    - Tool selection for this task stage
    """
    messages = []
    iteration = 0

    # Inject the initial task
    messages.append({"role": "user", "content": task})

    while iteration < max_iterations:
        # Apply context policy before each call
        # This is where context engineering happens
        active_messages = context_policy.prepare_messages(messages)
        active_tools = context_policy.select_tools(tools, messages)

        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=8096,
            system=system_prompt,
            tools=active_tools,
            messages=active_messages,
        )

        # Check stop conditions
        if response.stop_reason == "end_turn":
            return extract_final_answer(response)

        if response.stop_reason == "tool_use":
            # Execute tool calls and append results
            tool_results = execute_tool_calls(response.content)
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})

            # Apply backpressure signals (lint checks, test results, etc.)
            if backpressure := context_policy.check_backpressure(messages):
                messages.append({"role": "user", "content": backpressure})

        iteration += 1

    raise RuntimeError(f"Agent exceeded max_iterations={max_iterations}")


def extract_final_answer(response) -> str:
    """Extract text from the final response."""
    for block in response.content:
        if block.type == "text":
            return block.text
    return ""




def dispatch_tool(tool_name: str, tool_input: dict) -> str:
    """
    Route tool calls to their implementations.
    Replace with your actual tool registry in production.

    Example:
        tool_registry = {
            "read_file": read_file_tool,
            "write_file": write_file_tool, 
            "run_bash": run_bash_tool,
            "search_web": search_web_tool,
        }
        return tool_registry[tool_name](**tool_input)
    """
    raise NotImplementedError(f"Tool '{tool_name}' not registered. Add it to your tool registry.")

def execute_tool_calls(content_blocks: list) -> list[dict]:
    """
    Execute tool calls and format results.
    Returns a list of tool_result blocks for the next message.
    """
    results = []
    for block in content_blocks:
        if block.type == "tool_use":
            # Dispatch to actual tool implementation
            tool_output = dispatch_tool(block.name, block.input)
            results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": str(tool_output),
            })
    return results

from dataclasses import dataclass, field
from typing import Callable
import tiktoken

@dataclass
class ContextPolicy:
    """
    Governs context window management for a running agent.

    Design parameters:
    - model_context_limit: Total token limit for the model
    - compaction_threshold: Trigger compaction at this fraction of the window
    - system_prompt_budget: Max tokens reserved for system prompt
    - tool_budget: Max tokens reserved for tool schemas
    - task_headroom: Min tokens always kept free for the current task step
    """
    model_context_limit: int = 128_000
    compaction_threshold: float = 0.60       # Compact at 60% full
    system_prompt_budget: int = 6_000
    tool_budget: int = 20_000
    task_headroom: int = 16_000

    # Tool selector: given message history, return relevant tool subset
    tool_selector: Callable = field(default_factory=lambda: lambda tools, msgs: tools)

    # Compaction function: given messages, return a compressed version
    compactor: Callable = field(default_factory=lambda: default_compactor)

    def _count_tokens(self, messages: list[dict]) -> int:
        """Approximate token count for message list."""
        enc = tiktoken.get_encoding("cl100k_base")
        total = 0
        for msg in messages:
            if isinstance(msg["content"], str):
                total += len(enc.encode(msg["content"]))
            elif isinstance(msg["content"], list):
                for block in msg["content"]:
                    if isinstance(block, dict) and "text" in block:
                        total += len(enc.encode(block["text"]))
        return total

    @property
    def available_for_history(self) -> int:
        """Token budget remaining for conversation history."""
        return (
            self.model_context_limit
            - self.system_prompt_budget
            - self.tool_budget
            - self.task_headroom
        )

    @property
    def compaction_trigger(self) -> int:
        """Token count at which to trigger compaction."""
        return int(self.model_context_limit * self.compaction_threshold)

    def prepare_messages(self, messages: list[dict]) -> list[dict]:
        """
        Apply context policy to message history before each model call.

        Strategy:
        1. Check if current history exceeds compaction threshold
        2. If so, compact the oldest N messages into a summary
        3. Always preserve the most recent N turns verbatim (recency bias)
        4. Ensure task headroom is available
        """
        current_tokens = self._count_tokens(messages)

        if current_tokens > self.compaction_trigger:
            messages = self.compactor(messages, target_tokens=self.available_for_history)

        return messages

    def select_tools(self, all_tools: list[dict], messages: list[dict]) -> list[dict]:
        """
        Return only the tools relevant to the current task stage.
        Reduces tool-schema token overhead by 40-70% on typical agents.
        """
        return self.tool_selector(all_tools, messages)

    def check_backpressure(self, messages: list[dict]) -> str | None:
        """
        Check if backpressure signals should be injected.
        Returns a message to inject if action is needed, else None.

        Example signals: lint failures, test failures, build errors.
        These are injected as user messages to create correction loops.
        """
        # Implementation is domain-specific
        # A coding agent might run: subprocess.run(["tsc", "--noEmit"])
        return None


def default_compactor(messages: list[dict], target_tokens: int) -> list[dict]:
    """
    Default compaction: summarize oldest half, keep newest half verbatim.

    For production use, replace with an LLM-based summarizer:
    summary = llm.summarize(old_messages)
    """
    if len(messages) < 4:
        return messages

    split = len(messages) // 2
    old_messages = messages[:split]
    recent_messages = messages[split:]

    # In production: call an LLM to summarize old_messages
    summary_text = f"[COMPACTED: {len(old_messages)} earlier messages summarized. Key context: task in progress, {len(old_messages)} tool calls executed.]"

    return [
        {"role": "user", "content": summary_text},
        {"role": "assistant", "content": "Understood. Continuing from the compacted context."},
        *recent_messages,
    ]

import anthropic

compaction_client = anthropic.Anthropic()

def llm_compactor(messages: list[dict], target_tokens: int) -> list[dict]:
    """
    LLM-based compaction: summarize the first half of message history.

    This preserves semantic meaning better than truncation,
    and captures tool call patterns that matter for task continuation.

    Args:
        messages: Full message history
        target_tokens: Target token count for the compressed history

    Returns:
        Compressed message list with summary prepended
    """
    # Always preserve the most recent 20% of messages verbatim
    recency_cutoff = max(4, len(messages) - len(messages) // 5)
    messages_to_compact = messages[:recency_cutoff]
    messages_to_keep = messages[recency_cutoff:]

    # Format the old messages for the summarizer
    formatted_history = format_messages_for_summary(messages_to_compact)

    summary_response = compaction_client.messages.create(
        model="claude-haiku-4-5",  # Use a cheaper/faster model for compaction
        max_tokens=2048,
        system="""You are a context compression assistant. 
        Summarize the provided agent conversation history into a dense, 
        information-rich summary. Preserve:
        - What task the agent is working on
        - What has been completed (with key outputs/results)  
        - What tools were called and their outcomes
        - Any errors encountered and how they were resolved
        - The current state and what the agent should do next

        Be dense. Omit filler. Preserve all concrete artifacts, file paths,
        function names, and decision points that the agent will need.""",
        messages=[
            {
                "role": "user",
                "content": f"Summarize this agent conversation history:\n\n{formatted_history}"
            }
        ]
    )

    summary_text = summary_response.content[0].text

    # Construct the compressed history
    return [
        {
            "role": "user",
            "content": f"[CONTEXT SUMMARY - {len(messages_to_compact)} messages compacted]\n\n{summary_text}"
        },
        {
            "role": "assistant",
            "content": "Context loaded. I understand the task state and will continue from where the summary left off."
        },
        *messages_to_keep  # Keep recent messages verbatim
    ]


def format_messages_for_summary(messages: list[dict]) -> str:
    """Format message list into a readable string for the summarizer."""
    lines = []
    for msg in messages:
        role = msg["role"].upper()
        if isinstance(msg["content"], str):
            lines.append(f"[{role}]: {msg['content'][:500]}")
        elif isinstance(msg["content"], list):
            for block in msg["content"]:
                if isinstance(block, dict):
                    if block.get("type") == "text":
                        lines.append(f"[{role}]: {block['text'][:500]}")
                    elif block.get("type") == "tool_use":
                        lines.append(f"[{role} TOOL CALL]: {block['name']}({str(block['input'])[:200]})")
                    elif block.get("type") == "tool_result":
                        lines.append(f"[TOOL RESULT]: {str(block.get('content', ''))[:300]}")
    return "\n".join(lines)

# HANDOFF.md — Agent Context Reset

## Task Summary
[One paragraph: what we're building, what success looks like]

## Completed Steps
- [x] Step 1: Set up project structure (output: /src/index.ts created)
- [x] Step 2: Implemented auth module (output: /src/auth/*.ts, 847 lines)
- [x] Step 3: Database schema defined (output: schema.prisma, 23 models)

## Current Step
- [ ] Step 4: Implement REST API endpoints for /users, /posts, /comments

## Key Decisions Made
- Chose Hono over Express for the API layer (reason: better TypeScript types)
- Using Zod for request validation (aligned with schema.prisma types)
- JWT expiry set to 15min access / 7d refresh (security requirement from AGENTS.md)

## Files Written (do NOT overwrite)
/src/auth/jwt.ts, /src/auth/middleware.ts, /src/db/client.ts, schema.prisma

## Errors Encountered and Resolved
- TypeScript error in auth.ts: fixed by adding explicit return type annotations
- Prisma connection issue: resolved by adding DATABASE_URL to .env

## Next Actions
1. Read /src/auth/middleware.ts to understand the auth patterns in use
2. Implement GET /users/:id endpoint in /src/routes/users.ts
3. Run `npm run test:api` after each endpoint to verify

## Environment
- Node 22, TypeScript 5.8, Prisma 6.x, Hono 4.x
- Tests: `npm test` | Lint: `npm run lint` | Build: `npm run build`

import anthropic
import json

client = anthropic.Anthropic()

def run_planning_agent(task: str) -> dict:
    """
    Planner: Decomposes the task and writes a structured execution plan.
    Returns a plan dict that the generator will follow.
    """
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=4096,
        system="""You are a technical planning agent. Your job is to:
        1. Decompose the given task into concrete, verifiable sub-tasks
        2. Identify success criteria for each sub-task
        3. Flag potential risks or ambiguities
        4. Output a structured JSON plan

        Be specific. Vague plans produce vague outputs.""",
        messages=[{
            "role": "user",
            "content": f"Create an execution plan for: {task}"
        }]
    )

    # Extract JSON plan from response
    plan_text = response.content[0].text
    # In production: use structured output / tool_use to enforce JSON
    return {"plan": plan_text, "task": task}


def run_generator_agent(plan: dict, tools: list[dict]) -> dict:
    """
    Generator: Executes the plan using available tools.
    Produces artifacts (code, files, data) and a generation report.
    """
    generation_prompt = f"""
Execute the following plan:

{plan['plan']}

Original task: {plan['task']}

Use the available tools to complete each step. 
Write progress to PROGRESS.md after each major step.
When complete, output a JSON summary with:
- completed_steps: list of completed items
- artifacts: list of files/outputs created  
- issues_encountered: any problems found
- confidence_score: your self-assessment (0-10)
"""

    messages = [{"role": "user", "content": generation_prompt}]
    artifacts = {}

    # Run the generator loop (simplified — production version has full ContextPolicy)
    for _ in range(100):  # max iterations
        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=8096,
            tools=tools,
            messages=messages,
        )

        if response.stop_reason == "end_turn":
            final_output = extract_text(response)
            return {"output": final_output, "artifacts": artifacts, "plan": plan}

        if response.stop_reason == "tool_use":
            tool_results = execute_tools_and_collect_artifacts(response.content, artifacts)
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})

    return {"output": "max_iterations_reached", "artifacts": artifacts, "plan": plan}


def run_evaluator_agent(generation_result: dict, evaluation_criteria: list[str]) -> dict:
    """
    Evaluator: Critically assesses the generator's output.

    Key design choice: this agent is tuned to be skeptical.
    It does NOT have access to the generation history — 
    only the artifacts and evaluation criteria.
    """
    criteria_text = "\n".join(f"- {c}" for c in evaluation_criteria)

    eval_prompt = f"""
You are a rigorous code reviewer and technical evaluator. 
Your job is to critically assess the following output against specific criteria.

IMPORTANT: Be skeptical. Do not give credit for work that is incomplete, 
incorrect, or meets the criteria only superficially. When in doubt, fail the criterion.

Evaluation Criteria:
{criteria_text}

Output to evaluate:
{generation_result['output']}

Artifacts produced: {list(generation_result['artifacts'].keys())}

For each criterion, provide:
- PASS or FAIL
- Specific evidence for your judgment (quote relevant code/output)
- If FAIL: specific remediation steps

Output as JSON: {{"criteria_results": [...], "overall_pass": bool, "revision_signals": [...]}}
"""

    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=4096,
        system="You are a skeptical technical evaluator. Your bias is to find problems, not to validate work. A false negative (missing a real problem) is worse than a false positive (flagging non-problems).",
        messages=[{"role": "user", "content": eval_prompt}]
    )

    eval_text = response.content[0].text
    return {"evaluation": eval_text, "generation_result": generation_result}


def run_full_pipeline(
    task: str,
    tools: list[dict],
    evaluation_criteria: list[str],
    max_revision_rounds: int = 3
) -> dict:
    """
    Full Planner → Generator → Evaluator pipeline with revision loop.
    """
    # Phase 1: Planning
    plan = run_planning_agent(task)

    revision_round = 0
    while revision_round < max_revision_rounds:
        # Phase 2: Generation
        generation_result = run_generator_agent(plan, tools)

        # Phase 3: Evaluation
        eval_result = run_evaluator_agent(generation_result, evaluation_criteria)

        # Parse evaluation to check if we're done
        if is_evaluation_passing(eval_result["evaluation"]):
            return {
                "status": "success",
                "output": generation_result["output"],
                "artifacts": generation_result["artifacts"],
                "evaluation": eval_result["evaluation"],
                "revision_rounds": revision_round,
            }

        # Inject revision signals back into the plan for the next generator run
        revision_signals = extract_revision_signals(eval_result["evaluation"])
        plan["revision_notes"] = revision_signals
        plan["revision_round"] = revision_round + 1
        revision_round += 1

    return {
        "status": "max_revisions_reached",
        "output": generation_result["output"],
        "evaluation": eval_result["evaluation"],
    }


def is_evaluation_passing(evaluation_text: str) -> bool:
    """Parse the evaluator's JSON output and check overall_pass."""
    try:
        # In production: use structured output from the evaluator
        import re
        match = re.search(r'"overall_pass":\s*(true|false)', evaluation_text)
        return match and match.group(1) == "true"
    except Exception:
        return False


def extract_revision_signals(evaluation_text: str) -> list[str]:
    """Extract actionable revision signals from evaluator output."""
    # In production: parse the JSON revision_signals array
    return ["Review evaluator feedback and address all FAIL criteria before resubmitting."]


def extract_text(response) -> str:
    for block in response.content:
        if hasattr(block, 'text'):
            return block.text
    return ""


def execute_tools_and_collect_artifacts(content_blocks, artifacts: dict) -> list[dict]:
    """Execute tools, collect file artifacts, return tool results."""
    results = []
    for block in content_blocks:
        if hasattr(block, 'type') and block.type == "tool_use":
            output = dispatch_tool(block.name, block.input)  # dispatch to tool registry
            # Track file writes as artifacts
            if block.name in ("write_file", "create_file"):
                artifacts[block.input.get("path", "unknown")] = True
            results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": str(output),
            })
    return results

# AGENTS.md — Backend Coding Agent

## Identity
You are a senior backend engineer working on a Node.js/TypeScript API.
Stack: Node 22, TypeScript 5.8, Prisma 6, Hono 4, PostgreSQL 16.

## Core Constraints (Do NOT violate these)

### Safety
- NEVER run destructive database commands (DROP, TRUNCATE, DELETE without WHERE) 
  without explicit confirmation in the task spec.
  [INCIDENT: Agent dropped test table in production-connected environment, 2026-02-14]

- NEVER commit changes to main or master. Always use feature branches.
  [INCIDENT: Direct push to main caused CI failure, 2026-01-22]

- NEVER expose secrets, API keys, or credentials in code, logs, or output.
  [STANDARD: Security baseline]

### Code Quality  
- NEVER comment out tests. If a test is wrong, fix it or delete it.
  [INCIDENT: Commented-out tests reached production, undetected for 3 weeks]

- ALWAYS run `npm run lint` and `npm run build` before declaring a task complete.
  [INCIDENT: Type errors in production 4 separate times before this rule was added]

- ALWAYS add JSDoc to public functions. Private functions optional.
  [TEAM STANDARD: Code review surfaced this as consistent gap, 2026-03-01]

### Architecture
- Use Zod for ALL request/response validation. Do not use manual if-checks.
  [ARCHITECTURAL DECISION: Adopted 2025-11-15 for consistency]

- Repository pattern for DB access. Never write raw SQL in route handlers.
  [ARCHITECTURAL DECISION: Established 2025-09-01]

- Error responses MUST follow the ApiError schema in /src/types/errors.ts
  [INCIDENT: Inconsistent error formats broke mobile app, 2026-01-08]

## Workflow
1. Before starting: read PLAN.md if present, read PROGRESS.md if present
2. Write to PROGRESS.md after completing each major step
3. After completing: run full test suite, lint, and type-check before reporting done
4. If blocked by an error for > 3 attempts: write the error to BLOCKERS.md and stop

## Tool Use Priorities
1. Prefer reading existing files before writing new ones (understand before building)
2. Prefer modifying existing patterns over creating new ones
3. When writing tests: co-locate with source files (*.test.ts next to *.ts)

from mcp.server.fastmcp import FastMCP
from pathlib import Path
import subprocess

mcp = FastMCP("project-context-server")

@mcp.tool()
def get_project_structure(max_depth: int = 3) -> str:
    """
    Returns the project file structure up to max_depth.
    Use this at the start of any coding task to understand the codebase layout.
    """
    result = subprocess.run(
        ["find", ".", "-maxdepth", str(max_depth), "-not", "-path", "*/node_modules/*",
         "-not", "-path", "*/.git/*"],
        capture_output=True, text=True
    )
    return result.stdout


@mcp.tool()
def get_recent_git_changes(n_commits: int = 10) -> str:
    """
    Returns the last N commits with their diffs.
    Use this to understand what has changed recently and avoid regressions.
    """
    result = subprocess.run(
        ["git", "log", f"-{n_commits}", "--oneline", "--stat"],
        capture_output=True, text=True
    )
    return result.stdout


@mcp.resource("project://agents-config")
def get_agents_config() -> str:
    """Returns the current AGENTS.md content as a resource."""
    agents_md = Path("AGENTS.md")
    if agents_md.exists():
        return agents_md.read_text()
    return "No AGENTS.md found."


@mcp.tool()
def search_codebase(query: str, file_pattern: str = "*.ts") -> str:
    """
    Semantic search over the codebase using ripgrep.

    Args:
        query: Text to search for (supports regex)
        file_pattern: Glob pattern to limit search scope

    Returns:
        Matching lines with file paths and line numbers
    """
    result = subprocess.run(
        ["rg", "--type-add", f"custom:{file_pattern}", 
         "-t", "custom", "-n", "--context", "2", query],
        capture_output=True, text=True, cwd="."
    )
    # Limit output to avoid token overload
    lines = result.stdout.split("\n")[:100]
    return "\n".join(lines)


if __name__ == "__main__":
    mcp.run(transport="stdio")