# Build a single agent

Agent Mission Control developer docs > Microsoft Agent Framework (Python)

> Create an Agent Framework agent backed by FoundryChatClient, with function tools, MCP servers, structured output and streaming.

**Before you start:** A Foundry project endpoint, a chat model deployment name, and an az login session.

A single agent is one `Agent` object. Foundry hosts it directly, and the whole run is one model turn with whatever tools the agent calls. These examples target `agent-framework-core` 1.11.0.

## Install

`requirements.txt`
```text
agent-framework-core==1.11.0
agent-framework-foundry==1.10.1
agent-framework-foundry-hosting==1.0.0a260709
azure-ai-projects>=2.3.0
azure-identity>=1.19.0
```

## Configure

| Variable | Set by | Value |
| --- | --- | --- |
| `FOUNDRY_PROJECT_ENDPOINT` | You locally; the Foundry hosting runtime when deployed | `https://<resource>.services.ai.azure.com/api/projects/<project>` |
| `FOUNDRY_MODEL` | You, for local runs | `<model-deployment-name>` |
| `FOUNDRY_MODEL_NAME` or `AZURE_AI_MODEL_DEPLOYMENT_NAME` | The Foundry hosting runtime | The model deployment the hosted agent uses |
| `AZURE_TOKEN_CREDENTIALS` | You, in production | `prod`, to limit `DefaultAzureCredential` to production-safe credentials |

Authenticate with `DefaultAzureCredential`. It uses your `az login` session locally and the managed identity in Azure, with no code change. Avoid connection strings and account keys: they bypass Entra audit and rotation. Wrap clients in `with` or `async with` blocks so transports and token caches are released.

Read the model name from any of the three variables, so the same code runs locally and hosted:

`agent.py`
```python
import os


def _model() -> str:
    for var in ("FOUNDRY_MODEL", "FOUNDRY_MODEL_NAME", "AZURE_AI_MODEL_DEPLOYMENT_NAME"):
        value = os.getenv(var)
        if value:
            return value
    raise KeyError(
        "No model deployment configured: set FOUNDRY_MODEL, "
        "FOUNDRY_MODEL_NAME or AZURE_AI_MODEL_DEPLOYMENT_NAME."
    )
```

## Create the agent

`agent.py`
```python
import os
from typing import Annotated

from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from azure.identity import DefaultAzureCredential
from pydantic import BaseModel, Field

AGENT_NAME = "contoso-orders-agent"


def get_order_status(
    order_id: Annotated[str, Field(description="The order number to look up")],
) -> str:
    """Return the current status of an order."""
    return f"Order {order_id}: shipped"


class Answer(BaseModel):
    status: str
    message: str


def create_agent() -> Agent:
    client = FoundryChatClient(
        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
        model=_model(),
        credential=DefaultAzureCredential(),
    )
    return Agent(
        client=client,
        name=AGENT_NAME,
        instructions="You answer order-status questions. Always call the tool. Never guess.",
        tools=[get_order_status],
        default_options={"response_format": Answer, "store": False},
    )
```

- **Function tools:** pass plain functions (sync or `async def`) in `tools=`. The framework converts them and runs them in-process during the turn. Describe parameters with `Annotated[type, Field(description=...)]` and the tool with its docstring. Don't pass tool names as strings.
- **Structured output:** the Responses host does not forward a caller's `response_format`. Bake it into `default_options` so every run returns your model as JSON.
- **`store: False`:** the hosting layer manages conversation history. This is the usual setting for an agent AMC triggers.
- **Name:** keep `agent` in a single agent's name. AMC uses the name to classify what it imports; see [Register and test in AMC](https://www.agentmissioncontrol.ai/developers/af-register-and-test/).

## Run it locally

`run_local.py`
```python
import asyncio

from agent import Answer, create_agent


async def main() -> None:
    agent = create_agent()
    response = await agent.run("What is the status of order 1042?")
    answer = Answer.model_validate_json(response.text)
    print(answer.message)
    usage = response.usage_details or {}
    print(usage.get("input_token_count"), usage.get("output_token_count"))


asyncio.run(main())
```

> **Important:** `usage_details` is a `UsageDetails` TypedDict, which is a plain dict at runtime. Read it with `.get`. `getattr` on a dict always misses and silently returns nothing, which shows up in AMC as zero tokens.

### Streaming

Pass `stream=True` and iterate the updates with `async for`. Each update is an `AgentResponseUpdate`.

```python
async for update in agent.run("Summarize today's late orders", stream=True):
    if update.text:
        print(update.text, end="", flush=True)
```

A single update rarely carries the full token total. To total usage across a stream, fold each content's `usage_details` together with `add_usage_details`, as the token marker in [Harden an agent or workflow for AMC](https://www.agentmissioncontrol.ai/developers/af-amc-hardening/) does.

## Give a message-less trigger something to act on

AMC can start an agent with no meaningful input. Agent middleware runs on every hosted turn, so it can add a kickoff message before the model call. This one stamps the current time:

`agent.py`
```python
from collections.abc import Awaitable, Callable
from datetime import datetime, timezone

from agent_framework import AgentContext, Message, agent_middleware


@agent_middleware
async def inject_datetime(context: AgentContext, call_next: Callable[[], Awaitable[None]]) -> None:
    now = datetime.now(timezone.utc).isoformat()
    # Message takes contents (Content or str), not a text keyword.
    context.messages.append(Message(role="user", contents=[f"The current date and time is {now}."]))
    await call_next()

# Agent(..., middleware=[inject_datetime])
```

## Hosted tools and MCP servers

The skill documents these tool types, all imported from `agent_framework` and passed in `tools=` like a function tool:

| Tool | Use it for | Rules |
| --- | --- | --- |
| `HostedCodeInterpreterTool` | Running Python on the service: maths, data analysis, file processing | Give it uploaded files with `inputs=[HostedFileContent(file_id=...)]`. |
| `HostedFileSearchTool` | Searching documents in vector stores | Needs `inputs=[HostedVectorStoreContent(vector_store_id=...)]`. An empty `inputs` list is an error. Optional `max_results`. |
| `HostedWebSearchTool` | Current web information through Bing grounding | Needs a Bing connection, set as `BING_CONNECTION_ID`. |
| `HostedMCPTool` | An MCP server the service connects to | Configure `approval_mode`, `allowed_tools`, and `headers`. No context manager. |
| `MCPStreamableHTTPTool` | An MCP server your code connects to | Must be opened with `async with`. Set auth headers through a custom `http_client`. |

> **Important:** The skill wrote its hosted-tool and MCP examples against the earlier SDK surface and says to adjust class names for 1.11.0. Check that each class imports from your pinned version before you depend on it.

```python
from agent_framework import HostedMCPTool, MCPStreamableHTTPTool

# Service-managed: Foundry connects to the server.
docs_mcp = HostedMCPTool(
    name="Contoso docs",
    url="https://<your-mcp-server>/mcp",
    approval_mode="never_require",
    allowed_tools=["search_docs", "read_doc"],
)

# Client-managed: your process holds the connection, so open it first.
async with MCPStreamableHTTPTool(
    name="Contoso internal API",
    url="https://<your-internal-mcp-server>/mcp",
) as internal_mcp:
    agent = Agent(client=client, name=AGENT_NAME,
                  instructions="Use the docs and the internal API.",
                  tools=[docs_mcp, internal_mcp])
    response = await agent.run("What is our returns window?")
```

| `approval_mode` | Effect |
| --- | --- |
| `"never_require"` | Tool calls run without approval. |
| `"always_require"` | Every call needs approval. |
| `{"always_require_approval": [...], "never_require_approval": [...]}` | Per-tool lists. |

Use `HostedMCPTool` when the server is public and Foundry can reach it. Use `MCPStreamableHTTPTool` when the server is private or needs authentication, or when you want explicit control of the connection.

> **Note:** An unattended run can't complete an OAuth consent flow. If an MCP server only supports OAuth, call the service's REST API from a function tool instead, with a static credential read from an environment variable.

## Conversation state

On the hosted path the hosting layer keeps conversation history, which is why the agent sets `store: False`. A client that calls the hosted agent directly opens a session and passes its id on each call; see [Host and deploy on Foundry](https://www.agentmissioncontrol.ai/developers/af-host-and-deploy/). The skill's multi-turn thread examples (`get_new_thread()`, `thread=`) come from the earlier SDK surface, so confirm the thread API in your pinned version before you build on it.

Next: make the agent report tokens to AMC. See [Harden an agent or workflow for AMC](https://www.agentmissioncontrol.ai/developers/af-amc-hardening/).

## Related

- [Harden an agent or workflow for AMC](https://www.agentmissioncontrol.ai/developers/af-amc-hardening/)
- [Host and deploy on Foundry](https://www.agentmissioncontrol.ai/developers/af-host-and-deploy/)
- [Build a workflow](https://www.agentmissioncontrol.ai/developers/af-build-a-workflow/)

---
Source: https://www.agentmissioncontrol.ai/developers/af-build-an-agent/ · Last updated 2026-10-21 · Full index: https://www.agentmissioncontrol.ai/llms.txt