Agent Harness Explained: Build Production-Ready AI Agents with Microsoft Agent Framework

Meta Description: Learn what an agent harness is, why it matters for production AI systems, and how to implement one step-by-step using Microsoft Agent Framework's create_harness_agent -- with real Python code, architecture diagrams, and deep technical walkthroughs.

---

Table of Contents

1. Introduction -- The Hidden Complexity Tax
2. What is an Agent Harness?
   • The Manual Wiring Problem
   • Harness vs. DIY Agent
3. Microsoft Agent Framework Deep Dive
   • Architecture Overview
   • The Package Ecosystem
4. Anatomy of create_harness_agent
   • Function Invocation Layer
   • History Persistence
   • Compaction
   • TodoProvider
   • AgentModeProvider -- Plan/Execute Workflow
   • MemoryContextProvider
   • SkillsProvider
   • OpenTelemetry Integration
5. Full Implementation Walkthrough
   • Prerequisites & Installation
   • Minimal Harness Agent
   • Full Research Agent -- Line by Line
   • Streaming & Tool Observation
   • Customizing & Disabling Features
6. Running & Testing the Agent
7. Production Considerations
8. Conclusion

---

1. Introduction -- The Hidden Complexity Tax

You have a capable LLM. You have a clear use case. You write a chat loop in 20 lines of Python and it works -- until the context window fills up and the agent loses the thread. Or it calls a tool but forgets the result two turns later. Or it starts three things at once and has no way to track which are done. Or it crashes in production with no trace of what went wrong.

This is the hidden complexity tax of AI agents. Every production-grade agent needs: a tool-calling loop, conversation history management, context-window compaction, a planning mechanism, durable memory, skill extensibility, and observability. If you wire each of these yourself, you spend more time on infrastructure than on the actual intelligence. And when one breaks, the entire agent fails in ways that are nearly impossible to debug.

The agent harness pattern solves this by pre-assembling all of those components into a single, tested, configurable pipeline -- so you can focus on what your agent does, not how to keep it running.

In this deep dive, you'll learn:

• Exactly what an agent harness is and the problem it was designed to eliminate
• How Microsoft Agent Framework (MAF) implements the harness pattern with create_harness_agent
• A component-by-component breakdown of every subsystem the harness assembles
• A complete, production-ready Research Agent implementation with real Python code from the official MAF repository

Let's build something that lasts.

---

2. What is an Agent Harness?

The term "harness" comes from two established software patterns. In test engineering, a test harness is the scaffolding that configures, instruments, and tears down a system under test -- so test authors write test logic, not setup code. In dependency injection, a DI container is a wiring harness -- it resolves and connects components so application code never calls new directly.

An agent harness applies the same thinking to AI agents. It is:

A factory or container that constructs a fully wired, ready-to-run agent by assembling all required infrastructure components -- history, tools, memory, observability, planning -- from a single, declarative configuration point.

The key insight is separation of concerns: your agent instructions define what the agent should do; the harness defines how that intent is reliably executed, persisted, and observed.

Architecture Note: Think of the harness like a production-grade Kubernetes deployment vs. running a container manually with docker run. Both run your code, but one handles restarts, resource limits, networking, logging, and scaling automatically.

2.1 The Manual Wiring Problem

To appreciate what the harness eliminates, consider what you would have to build manually for a real agent:

Component	Manual Responsibility
Tool Calling Loop	Detect tool calls, dispatch, collect results, re-invoke model, set termination condition
History Management	Serialize/deserialize conversation history, decide storage backend, handle multi-turn sessions
Context Compaction	Monitor token count, decide eviction strategy, preserve semantic context
Planning / Todo	Design a task-tracking schema, prompt the model to use it, parse structured outputs
Mode Management	Track agent state (planning vs. executing), implement approval gates
Persistent Memory	Define memory schema, write/read from durable store, inject into context at the right time
Skills Loading	Design skill discovery protocol, filter by relevance, progressively inject into context
Telemetry	Instrument every model call, tool dispatch, and context mutation with spans and metrics

Writing all of this correctly, testing it, and keeping it working as your LLM provider updates its API is a substantial engineering effort. The harness pattern packages this effort once, so you -- and every developer on your team -- never has to repeat it.

2.2 Harness vs. DIY Agent

The difference is immediately visible in code. Here is a minimal but realistic "manual" agent that handles just the tool-calling loop:

# ? DIY approach -- and this is just the tool loop, not memory, compaction, or telemetry
import json

async def run_agent_manually(client, tools, messages):
    while True:
        response = await client.chat(messages=messages, tools=tools)

        if response.finish_reason == "stop":
            return response.content

        if response.finish_reason == "tool_calls":
            messages.append(response.message)  # add assistant turn

            for tool_call in response.tool_calls:
                # Manually dispatch each tool
                tool_fn = tool_registry.get(tool_call.function.name)
                if not tool_fn:
                    raise ValueError(f"Unknown tool: {tool_call.function.name}")

                result = await tool_fn(**json.loads(tool_call.function.arguments))

                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": str(result)
                })
            # Loop back to model -- now you ALSO need token counting,
            # history management, todo tracking, telemetry...

Now here is the harness equivalent that does all of the above plus history, compaction, planning, memory, and telemetry:

# ? Harness approach -- full production pipeline in 4 lines
from agent_framework import create_harness_agent
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential

agent = create_harness_agent(
    client=FoundryChatClient(credential=AzureCliCredential()),
    max_context_window_tokens=128_000,
    max_output_tokens=16_384,
)

Same outcome. Orders of magnitude less surface area for bugs.

---

3. Microsoft Agent Framework Deep Dive

Microsoft Agent Framework (MAF) is Microsoft's open-source, production-grade framework for building AI agents and multi-agent workflows. It is the direct successor to both AutoGen and Semantic Kernel -- combining AutoGen's simple, composable agent abstractions with Semantic Kernel's enterprise features: session-based state management, type safety, middleware pipelines, and comprehensive telemetry.

MAF supports:

• Multi-language -- Full parity between Python and C#/.NET
• Multi-provider -- Azure AI Foundry, OpenAI, Azure OpenAI, Anthropic (Claude), Google Gemini, Amazon Bedrock, Mistral, Ollama, GitHub Copilot SDK, and more
• Multi-pattern -- Single agents, sequential workflows, concurrent workflows, handoff patterns, group-chat patterns, human-in-the-loop
• Multi-deployment -- Local dev, Azure Functions, Foundry Hosted Agents (containerized Micro VMs), Durable Task

3.1 Architecture Overview

At the center of MAF is the Agent base class. All agent types derive from this common base, which defines a consistent interface for multi-agent orchestration.

The default agent runtime execution model follows a deterministic loop:

User Message
    ?
    ?
???????????????????????????????????????????????????
?                  Agent Runtime                  ?
?                                                 ?
?  1. Context Assembly                            ?
?     ?? HistoryProvider + ContextProviders       ?
?         (Memory, Skills, Mode, Todos)           ?
?                                                 ?
?  2. Middleware Pre-Processing                   ?
?     ?? Telemetry, Compaction checks             ?
?                                                 ?
?  3. Model Inference                             ?
?     ?? FoundryChatClient / OpenAI / Anthropic   ?
?                                                 ?
?  4. Tool Dispatch Loop                          ?
?     ?? FunctionInvocationLayer                  ?
?         (Tool call -> execute -> result -> loop)  ?
?                                                 ?
?  5. History Persistence                         ?
?     ?? Saved after every service call           ?
?                                                 ?
?  6. Middleware Post-Processing                  ?
?     ?? Telemetry spans closed                   ?
???????????????????????????????????????????????????
    ?
    ?
Agent Response (streaming or complete)

This loop runs transparently on every agent.run() call. The harness ensures all six phases are populated and correctly ordered.

3.2 The Package Ecosystem

MAF's Python implementation is split into composable packages. The top-level agent-framework meta-package installs all of them:

pip install agent-framework

Package	Purpose
core	Agent base, session management, middleware, context providers
foundry	FoundryChatClient, Azure AI Foundry integration
tools	get_web_search_tool() and other built-in tool factories
orchestrations	Multi-agent workflow graphs (sequential, concurrent, handoff)
devui	Interactive browser-based DevUI for local agent debugging
a2a	Agent-to-Agent (A2A) protocol support for cross-platform agents
lab	Experimental features: benchmarking, RL, research

---

4. Anatomy of create_harness_agent

create_harness_agent is a factory function -- a single call that constructs and wires a fully operational agent. Let's look at its complete signature and then break down every component it assembles:

from agent_framework import create_harness_agent

agent = create_harness_agent(
    # Required: the LLM backend
    client=client,

    # Required: token budget for the agent's context window
    max_context_window_tokens=128_000,
    max_output_tokens=16_384,

    # Optional: agent identity
    name="MyAgent",
    description="What this agent does.",
    agent_instructions="System prompt / instructions for the agent.",

    # Optional: feature toggles (all enabled by default)
    disable_todo=False,           # TodoProvider
    disable_mode=False,           # AgentModeProvider
    disable_compaction=False,     # CompactionProvider
    memory_store=None,            # MemoryContextProvider (None = disabled)
    skills_directory=None,        # SkillsProvider (None = disabled)
    extra_tools=[],               # Additional tools to inject
)

This single call assembles 8 sub-systems automatically. Here is each one in depth.

4.1 Function Invocation Layer

This is the agent's agentic loop engine -- the component that makes the agent actually do things rather than just respond.

When a model returns a response containing tool calls, the FunctionInvocationLayer middleware:

1. Detects tool call requests in the model's response
2. Dispatches each call to the registered tool function
3. Collects results and serializes them back into message history
4. Re-invokes the model with updated history
5. Repeats until the model returns a final text response
6. Enforces a maximum iteration limit to prevent infinite loops

Tool functions are registered as plain Python callables with type-annotated signatures. The framework automatically generates JSON schema from the annotations:

async def search_database(query: str, limit: int = 10) -> list[dict]:
    """Search the product database for matching items.

    Args:
        query: The search query string.
        limit: Maximum number of results to return.

    Returns:
        List of matching product records.
    """
    # Your implementation here
    return results

# The framework auto-generates the JSON schema and handles dispatch
agent = create_harness_agent(
    client=client,
    max_context_window_tokens=128_000,
    max_output_tokens=16_384,
    extra_tools=[search_database],
)

4.2 History Persistence

The InMemoryHistoryProvider manages the conversation history for the agent's session. What makes this different from a plain message list is when persistence occurs: after every individual model service call, not just at the end of the agent's turn.

This matters for reliability. Consider an agent making three tool calls in a single turn. Without per-call persistence, a crash on the third call loses work from calls one and two. With per-call persistence, the agent can resume from the last successful state.

# Sessions are created per-conversation to isolate history
session = agent.create_session()

# All turns in this session share and accumulate history
result_1 = await agent.run("Research quantum computing trends", session=session)
result_2 = await agent.run("Now summarize your findings in bullet points", session=session)
# Turn 2 has full memory of everything from turn 1

For production deployments that require history to survive process restarts, MAF supports swapping InMemoryHistoryProvider with durable backends -- Azure Cosmos DB, Redis, or any custom implementation via the IHistoryProvider interface.

4.3 Compaction

Context-window overflow is one of the most common production failures for long-running agents. The CompactionProvider addresses this with a two-pronged strategy:

Sliding Window: When total token count approaches max_context_window_tokens, older messages are pruned. The system prompt and most recent N messages are always preserved.

Tool Result Compaction: Tool results are often verbose -- a web search might return thousands of tokens. The compaction layer intelligently summarizes tool results when under token pressure, keeping essential information while reducing usage.

# Token budget math:
# available_context = max_context_window_tokens - max_output_tokens - system_prompt_tokens
# Compaction fires BEFORE the hard limit is hit

agent = create_harness_agent(
    client=client,
    max_context_window_tokens=128_000,   # GPT-4o's total window
    max_output_tokens=16_384,            # Reserved for model's response
    # ~112K tokens available for history + context providers
    # Compaction fires automatically to stay within budget
)

# Or disable it if you manage token budget yourself:
agent = create_harness_agent(
    client=client,
    max_context_window_tokens=128_000,
    max_output_tokens=16_384,
    disable_compaction=True,
)

4.4 TodoProvider

The TodoProvider gives the agent a structured task-tracking system within its own context. Rather than relying on the agent to implicitly remember its progress, TodoProvider exposes tool functions the agent can call to manage an explicit todo list:

• create_todo(title, description) -- Creates a new work item
• complete_todo(id) -- Marks an item done
• list_todos() -- Gets all pending items
• update_todo(id, status) -- Updates item status

This is deceptively powerful. The agent can decompose a complex task into explicit work items, track completion, and resist skipping steps -- all driven by its own reasoning.

User: "Research the top 5 AI agent frameworks and compare them."

Agent (internal reasoning with TodoProvider):
  -> create_todo("Research AutoGen")
  -> create_todo("Research LangGraph")
  -> create_todo("Research CrewAI")
  -> create_todo("Research Microsoft Agent Framework")
  -> create_todo("Write comparison table")

  [Agent proceeds autonomously]

  -> complete_todo("Research AutoGen")
  -> complete_todo("Research LangGraph")
  -> ... and so on until all todos are done

4.5 AgentModeProvider -- Plan/Execute Workflow

The AgentModeProvider implements a two-phase workflow that mirrors how skilled professionals approach complex tasks:

Phase 1 -- Plan Mode (Interactive)

In plan mode the agent is collaborative:

• Asks the user clarifying questions to reduce ambiguity
• Creates a detailed todo list outlining its intended approach
• Presents the plan for human review
• Waits for explicit approval before proceeding

This is the human-in-the-loop gate -- users can modify the plan or redirect entirely before any autonomous work begins.

Phase 2 -- Execute Mode (Autonomous)

Once approved, the agent:

• Works through its todo list independently
• Makes tool calls, searches the web, writes to memory
• Marks todos complete as each is finished
• Streams progress to the user in real time
• Returns to plan mode if it encounters a blocking ambiguity

This pattern is what makes agents trustworthy in production -- the human sees and approves the plan, and the agent executes it faithfully.

4.6 MemoryContextProvider

The MemoryContextProvider provides file-based durable memory -- a persistent store the agent can read from and write to across sessions and compaction events.

This solves a critical gap in pure in-context memory: when compaction fires, older information is dropped. If the agent wrote a research finding 50 messages ago, it may no longer be in context. But if the agent saved it to memory storage, the MemoryContextProvider re-injects it at the start of every context assembly cycle.

import tempfile
from pathlib import Path

memory_dir = Path(tempfile.mkdtemp()) / "agent_memory"
memory_dir.mkdir(parents=True, exist_ok=True)

agent = create_harness_agent(
    client=client,
    max_context_window_tokens=128_000,
    max_output_tokens=16_384,
    memory_store=str(memory_dir),  # Enable file-based persistent memory
)

# The agent can now use internal memory tools:
# memory_write(key, content) -- Persist information to disk
# memory_read(key)           -- Retrieve persisted information
# Future sessions re-inject saved memory automatically into context

4.7 SkillsProvider

The SkillsProvider enables runtime skill discovery and progressive loading. Skills are domain-specific capability packages -- sets of tools, instructions, and knowledge -- that the agent can discover and load on demand.

Rather than loading all possible tools upfront (wasting context tokens), SkillsProvider implements progressive loading: it first presents skill summaries, and the agent selectively loads full skill definitions it actually needs. A general-purpose agent can become a specialized SQL expert, code reviewer, or API integration specialist -- dynamically, based on what the task requires.

Skills can be authored in three ways (as of the latest MAF release):

1. File-based -- YAML or JSON files in a skills directory
2. Code-based -- Python classes with annotated methods
3. Inline -- Defined programmatically at agent creation time

agent = create_harness_agent(
    client=client,
    max_context_window_tokens=128_000,
    max_output_tokens=16_384,
    skills_directory="./skills",  # Load skills from this directory
)

# Example skills/ directory:
# skills/
# ??? web_research.yaml     <- Web search + summarization capabilities
# ??? data_analysis.yaml    <- Pandas/SQL data analysis capabilities
# ??? code_review.yaml      <- Code quality review capabilities
# ??? report_writing.yaml   <- Document formatting capabilities

4.8 OpenTelemetry Integration

The AgentTelemetryLayer instruments every observable event in the agent's lifecycle:

Telemetry Event	What is Captured
agent.run.start	Session ID, user input, agent name, timestamp
agent.model.call	Model name, token counts (prompt/completion), latency
agent.tool.call	Tool name, arguments, result size, latency
agent.compaction.triggered	Token counts before/after, messages evicted
agent.mode.switch	From/to mode, trigger reason
agent.run.complete	Total tokens, tool calls, total latency, final status

# Configure an OTLP exporter before creating the agent
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor

# Point at Azure Monitor / Jaeger / Grafana Tempo / Datadog, etc.
exporter = OTLPSpanExporter(endpoint="http://localhost:4317")
provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)

# The harness agent automatically uses the configured tracer -- no extra code needed
agent = create_harness_agent(
    client=client,
    max_context_window_tokens=128_000,
    max_output_tokens=16_384,
)

---

5. Full Implementation Walkthrough

Now let's build the complete harness-based Research Agent from the official MAF repository -- line by line, with full commentary.

5.1 Prerequisites & Installation

# Step 1: Install Microsoft Agent Framework
pip install agent-framework

# Step 2: Authenticate with Azure
az login

# Step 3: Create your .env file
cat > .env << EOF
FOUNDRY_PROJECT_ENDPOINT=https://your-project.services.ai.azure.com/api/projects/your-project-name
FOUNDRY_MODEL=gpt-4o
EOF

Finding Your Foundry Endpoint: Navigate to your Azure AI Foundry project in the Azure Portal -> Project Overview -> copy the "Project endpoint" URL.

5.2 Minimal Harness Agent

Here is the absolute minimum viable harness agent -- 4 lines of setup that give you a fully operational agent with history, compaction, todos, planning, and telemetry:

# minimal_harness.py
import asyncio
from agent_framework import create_harness_agent
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential
from dotenv import load_dotenv

async def main():
    # MAF does NOT auto-load .env -- must be explicit
    load_dotenv()

    # 1. LLM backend using your az login session (no secrets in code)
    client = FoundryChatClient(credential=AzureCliCredential())

    # 2. Harness agent -- all 8 sub-systems active by default
    agent = create_harness_agent(
        client=client,
        max_context_window_tokens=128_000,
        max_output_tokens=16_384,
    )

    # 3. Create a session (isolates conversation history)
    session = agent.create_session()

    # 4. Run the agent
    response = await agent.run(
        "What are the latest trends in AI agent frameworks?",
        session=session,
    )
    print(response)

if __name__ == "__main__":
    asyncio.run(main())

5.3 Full Research Agent -- Line by Line

Now the complete research agent from the official repository with detailed inline commentary:

# harness_research.py
# Source: https://github.com/microsoft/agent-framework/tree/main/python/samples/02-agents/harness

import asyncio
from agent_framework import create_harness_agent
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential
from dotenv import load_dotenv

# ??????????????????????????????????????????????????????????????????????????
# SECTION 1: Agent Instructions
#
# The system prompt is injected at every context assembly cycle.
# It is ALWAYS present -- even after compaction fires -- because the harness
# preserves the system prompt unconditionally.
#
# Key best practices here:
#   - Clear role: "You are a research assistant"
#   - Quality criteria: multiple sources, cross-referencing
#   - Output format: Markdown, headings, inline citations
#   - Durable memory: save final report to file memory so it survives
#     compaction and persists across sessions
# ??????????????????????????????????????????????????????????????????????????

RESEARCH_INSTRUCTIONS = """\
## Research Assistant Instructions

You are a research assistant. When given a research topic, research it
thoroughly using web search and web browsing.
Use your knowledge to form good search queries and hypotheses, but always
verify claims with the tools available to you rather than relying on memory alone.

### Research quality

Consult multiple sources when possible and cross-reference key claims.
When sources disagree, note the discrepancy and explain which source you
consider more reliable and why.
If a web page fails to load or a search returns irrelevant results, try
alternative search queries or sources before moving on.
Track your sources -- you will need them when presenting results.

### Presenting results

When presenting your final findings:
- Use Markdown formatting for clarity.
- Use clear sections with headings for each major topic or sub-question.
- Cite your sources inline (e.g., "According to [source name](URL), ...").
- End with a brief summary of key takeaways.
- Save the final research report to file memory so it survives compaction
  and can be referenced in future sessions.
"""


async def main() -> None:
    # ?????????????????????????????????????????????????????????????????????
    # SECTION 2: Environment & Client Setup
    # ?????????????????????????????????????????????????????????????????????

    load_dotenv()  # Reads FOUNDRY_PROJECT_ENDPOINT and FOUNDRY_MODEL

    # AzureCliCredential uses your `az login` session -- ideal for dev.
    # For production: replace with ManagedIdentityCredential
    client = FoundryChatClient(credential=AzureCliCredential())

    # ?????????????????????????????????????????????????????????????????????
    # SECTION 3: Harness Agent Assembly
    # ?????????????????????????????????????????????????????????????????????

    agent = create_harness_agent(
        client=client,
        max_context_window_tokens=128_000,   # Total model context window
        max_output_tokens=16_384,            # Reserved for model's response
        name="ResearchAgent",
        description="A research assistant that plans and executes research tasks.",
        agent_instructions=RESEARCH_INSTRUCTIONS,
        # All features active by default:
        # - TodoProvider: tracks research tasks as explicit work items
        # - AgentModeProvider: plan/execute two-phase workflow
        # - CompactionProvider: sliding window + tool result compaction
        # - SkillsProvider: progressive skill discovery
        # - MemoryContextProvider: file-based durable memory
        # - AgentTelemetryLayer: full OpenTelemetry instrumentation
    )

    # ?????????????????????????????????????????????????????????????????????
    # SECTION 4: Session -- isolates this conversation's history and state
    # ?????????????????????????????????????????????????????????????????????

    session = agent.create_session()

    print("Research Assistant (powered by create_harness_agent)")
    print("=" * 50)
    print("Enter a research topic to get started.")
    print("Type /exit to end the session.\n")

    # ?????????????????????????????????????????????????????????????????????
    # SECTION 5: Interactive streaming chat loop
    # ?????????????????????????????????????????????????????????????????????

    while True:
        user_input = input("You: ").strip()
        if not user_input:
            continue
        if user_input.lower() == "/exit":
            print("\nGoodbye!")
            break

        print("\nAssistant: ", end="", flush=True)

        # agent.run(..., stream=True) returns AsyncGenerator[AgentUpdate, None]
        # Each AgentUpdate has:
        #   update.text     -- streaming text fragment from the model
        #   update.contents -- list of structured content items (tool calls, etc.)
        async for update in agent.run(user_input, session=session, stream=True):
            if update.contents:
                for content in update.contents:
                    if content.type == "function_call":
                        # Tool is being invoked -- show users what's happening
                        print(f"\n  [calling tool: {content.name}]", flush=True)
                        print("  ", end="", flush=True)

                    # Handle web search events from the built-in search tool
                    elif content.type in ("search_tool_call", "search_tool_result") and \
                         getattr(content, "tool_name", None) == "web_search":
                        action = None
                        if content.type == "search_tool_result" and isinstance(content.result, dict):
                            action = content.result.get("action", {})
                        elif content.type == "search_tool_call":
                            action = content.arguments if isinstance(content.arguments, dict) else None

                        if action:
                            action_type = action.get("type", "search")
                            if action_type == "search":
                                queries = action.get("queries") or []
                                query_str = ", ".join(f'"{q}"' for q in queries) \
                                            if queries else action.get("query", "")
                                print(f"\n  ? Web search: {query_str}", flush=True)
                                print("  ", end="", flush=True)
                            elif action_type == "open_page":
                                url = action.get("url", "(unknown)")
                                print(f"\n  ? Opening: {url}", flush=True)
                                print("  ", end="", flush=True)
                            elif action_type == "find_in_page":
                                pattern = action.get("pattern", "")
                                print(f'\n  ? Find in page: "{pattern}"', flush=True)
                                print("  ", end="", flush=True)
                            else:
                                print(f"\n  ? Web search: {action_type}", flush=True)
                                print("  ", end="", flush=True)

            # Stream text fragments as they arrive from the model
            if update.text:
                print(update.text, end="", flush=True)

        print("\n")


if __name__ == "__main__":
    asyncio.run(main())

5.4 Streaming & Tool Observation

Here is a clean, reusable streaming handler you can adapt for your own applications:

# streaming_handler.py -- Reusable streaming output handler

async def stream_agent_response(agent, user_input: str, session) -> str:
    """
    Stream an agent response, printing text as it arrives
    and logging tool calls with their arguments and results.

    Returns:
        The complete assembled response text.
    """
    full_response_parts: list[str] = []

    print("Assistant: ", end="", flush=True)

    async for update in agent.run(user_input, session=session, stream=True):

        # Stream text as it arrives
        if update.text:
            print(update.text, end="", flush=True)
            full_response_parts.append(update.text)

        # Handle structured content items
        if update.contents:
            for content in update.contents:

                if content.type == "function_call":
                    args_summary = str(content.arguments)[:100]
                    print(f"\n  ? Tool call: {content.name}({args_summary}...)", flush=True)

                elif content.type == "function_result":
                    result_preview = str(content.result)[:80]
                    print(f"\n  ? Result: {result_preview}...", flush=True)

                elif content.type == "search_tool_call":
                    if hasattr(content, "arguments") and isinstance(content.arguments, dict):
                        query = content.arguments.get("query", "")
                        print(f"\n  ? Searching: '{query}'", flush=True)

                elif content.type == "search_tool_result":
                    if isinstance(content.result, dict):
                        url = content.result.get("url", "")
                        if url:
                            print(f"\n  ? Retrieved: {url}", flush=True)

    print("\n")
    return "".join(full_response_parts)

5.5 Customizing & Disabling Features

# ?? Pattern 1: Lean agent (no planning, no todo management) ????????????
# Use for: Simple Q&A, single-turn tasks, low-latency scenarios

lean_agent = create_harness_agent(
    client=client,
    max_context_window_tokens=32_000,
    max_output_tokens=4_096,
    name="QuickAnswerAgent",
    agent_instructions="You are a concise Q&A assistant. Answer briefly and directly.",
    disable_todo=True,       # No task tracking for Q&A
    disable_mode=True,       # No plan/execute modes needed
    disable_compaction=False, # Keep compaction for long conversations
)

# ?? Pattern 2: Full research agent with persistent memory ???????????????
# Use for: Long-running research, multi-session workflows

from pathlib import Path

memory_path = Path("./research_memory")
memory_path.mkdir(exist_ok=True)

research_agent = create_harness_agent(
    client=client,
    max_context_window_tokens=128_000,
    max_output_tokens=16_384,
    name="ResearchAgent",
    agent_instructions=RESEARCH_INSTRUCTIONS,
    memory_store=str(memory_path),  # Enable file-based persistent memory
)

# ?? Pattern 3: Agent with custom enterprise tools ??????????????????????
# Use for: Domain-specific agents with proprietary data access

from agent_framework.tools import get_web_search_tool

async def query_internal_db(query: str, department: str = "all") -> list[dict]:
    """Query the internal company database.

    Args:
        query: Search query for the database.
        department: Filter by department name, or 'all' for global search.

    Returns:
        List of matching records.
    """
    # Your internal DB implementation
    return []

async def get_slack_messages(channel: str, days_back: int = 7) -> list[dict]:
    """Retrieve recent Slack messages from a channel.

    Args:
        channel: Slack channel name (without #).
        days_back: Number of days of history to retrieve.

    Returns:
        List of message objects with sender, timestamp, and text.
    """
    # Your Slack API implementation
    return []

custom_agent = create_harness_agent(
    client=client,
    max_context_window_tokens=128_000,
    max_output_tokens=16_384,
    name="InternalResearchAgent",
    agent_instructions="You are an internal research assistant with access to company data.",
    extra_tools=[
        get_web_search_tool(),   # Built-in web search
        query_internal_db,       # Custom: internal database
        get_slack_messages,      # Custom: Slack integration
    ],
)

---

6. Running & Testing the Agent

# Set environment variables
export FOUNDRY_PROJECT_ENDPOINT="https://your-project.services.ai.azure.com/api/projects/your-project-name"
export FOUNDRY_MODEL="gpt-4o"

# Authenticate with Azure
az login

# Run the research agent
python harness_research.py

Expected terminal output when you ask a research question:

Research Assistant (powered by create_harness_agent)
==================================================
Enter a research topic to get started.
Type /exit to end the session.

You: Research the current state of AI agent frameworks in 2026

Assistant:
  [calling tool: switch_to_plan_mode]
  [calling tool: create_todo]
  [calling tool: create_todo]
  [calling tool: create_todo]
  [calling tool: switch_to_execute_mode]

Here is my research plan. I will:
1. Survey the major frameworks (MAF, AutoGen, LangGraph, CrewAI, LlamaIndex)
2. Look up recent benchmarks and community activity
3. Compare feature sets in a table
4. Summarize key takeaways

Shall I proceed?

You: Yes, go ahead.

Assistant:
  ? Web search: "Microsoft Agent Framework 2026 features"
  ? Opening: https://github.com/microsoft/agent-framework
  ? Web search: "LangGraph vs AutoGen comparison 2026"
  [calling tool: complete_todo]
  ? Web search: "CrewAI production readiness 2026"
  ...

## AI Agent Frameworks -- State of the Ecosystem (2026)

### Microsoft Agent Framework (MAF)
According to [Microsoft DevBlogs](https://devblogs.microsoft.com/agent-framework/), MAF 1.0 ...

Pro Tip: Launch the built-in DevUI for a visual debugging experience during development: install agent-framework[devui] and run agent devui in your project directory.

---

7. Production Considerations

The harness pattern dramatically reduces the operational burden of running agents in production, but there are still architectural decisions to make:

Token Budget Sizing

The right max_context_window_tokens depends on your model and workload. A rule of thumb:

• Leave 15-20% headroom -- Don't set the value to the model's exact maximum. Tokenizers can be imprecise and system overheads vary.
• For research agents -- Use the model's full window (128K for GPT-4o) to allow thorough multi-source research.
• For Q&A agents -- 16K-32K is usually sufficient and reduces cost significantly.
• For coding agents -- 64K-128K is recommended to handle large file contexts.

Durable History Backends

For multi-user production deployments, InMemoryHistoryProvider is insufficient -- process restarts lose all history. MAF supports pluggable history backends:

# Example: using a custom durable history provider (pattern)
from agent_framework.core import Agent

# Implement IHistoryProvider backed by your chosen store
# (CosmosDB, Redis, PostgreSQL, etc.) and pass it to the agent builder
# See MAF documentation for the full IHistoryProvider interface

Security: Prompt Injection Defense

As of May 2026, MAF ships FIDES (Flow Integrity Deterministic Enforcement System) as a middleware -- the #1 defense against prompt injection (OWASP LLM Top 10 risk #1). FIDES assigns integrity labels (trusted/untrusted) and confidentiality labels (public/private) to every piece of content flowing through the agent. Labels propagate automatically, and policy enforcement is deterministic -- not heuristic.

# Enable FIDES middleware for production agents
from agent_framework.security import FidesMiddleware

agent = create_harness_agent(
    client=client,
    max_context_window_tokens=128_000,
    max_output_tokens=16_384,
    # Additional middleware can be composed with the harness
    # Consult MAF docs for middleware registration API
)

Deploying to Azure Foundry Hosted Agents

When you're ready to move from local development to production, Foundry Hosted Agents provides containerized Micro VM hosting with built-in identity, autoscaling, session state management, and versioning. The migration from a local harness agent is minimal:

# The agent code is identical -- only the hosting changes
# Add foundry_hosting package and declare your agent as a hosted endpoint
# See: https://github.com/microsoft/agent-framework/tree/main/python/samples/04-hosting

Observability in Production

With OpenTelemetry already wired in by the harness, you need only configure your exporter to get production-grade visibility:

• Azure Monitor Application Insights -- For teams already in the Azure ecosystem
• Grafana + Tempo -- For open-source observability stacks
• Datadog / Dynatrace -- For enterprise APM platforms

Key dashboards to build: token consumption per session, tool call latency distribution, compaction frequency, and agent error rates.

---

8. Conclusion

The agent harness pattern is the difference between an AI demo and a production AI system. It acknowledges a fundamental truth: building the intelligence of an agent is the easy part. Keeping that intelligence reliable, observable, durable, and safe under production load is the hard part -- and the harness handles that hard part for you.

Microsoft Agent Framework's create_harness_agent is the most complete open-source implementation of this pattern available today. In a single factory call it wires together eight battle-tested subsystems -- function invocation, history persistence, context compaction, todo-based planning, plan/execute mode management, durable file memory, progressive skill loading, and OpenTelemetry instrumentation -- all individually configurable, all working in concert.

Here is what to take away from this deep dive:

1. The harness is not magic -- it is a well-designed factory that assembles components you would otherwise wire manually, with all the ordering, configuration, and failure-handling done correctly
2. Start with the harness, strip down if needed -- it is always easier to disable features you don't need (disable_todo=True, disable_mode=True) than to add them later
3. The Plan/Execute pattern is the right default for any task-oriented agent -- it keeps humans in control while enabling genuine autonomy
4. Telemetry is non-negotiable in production -- the harness gives it to you for free, so configure an exporter before day one
5. The official GitHub sample is your north star -- harness_research.py is production-quality code worth reading end to end

Ready to build?

pip install agent-framework
az login
python harness_research.py

Explore the full framework at github.com/microsoft/agent-framework, join the community on Discord, and check the latest patterns on the official blog.

The infrastructure is handled. Go build the intelligence. ?

---

All code samples in this article are sourced from or based on the official Microsoft Agent Framework repository (MIT License). Verify all API signatures against the latest release before deploying to production.