Browse the docs

Build a single agent

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

Before you startA Foundry project endpoint, a chat model deployment name, and an az login session.View as Markdown

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
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

VariableSet byValue
FOUNDRY_PROJECT_ENDPOINTYou locally; the Foundry hosting runtime when deployedhttps://<resource>.services.ai.azure.com/api/projects/<project>
FOUNDRY_MODELYou, for local runs<model-deployment-name>
FOUNDRY_MODEL_NAME or AZURE_AI_MODEL_DEPLOYMENT_NAMEThe Foundry hosting runtimeThe model deployment the hosted agent uses
AZURE_TOKEN_CREDENTIALSYou, in productionprod, 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
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
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.

Run it locally

run_local.py
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())

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 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
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:

ToolUse it forRules
HostedCodeInterpreterToolRunning Python on the service: maths, data analysis, file processingGive it uploaded files with inputs=[HostedFileContent(file_id=...)].
HostedFileSearchToolSearching documents in vector storesNeeds inputs=[HostedVectorStoreContent(vector_store_id=...)]. An empty inputs list is an error. Optional max_results.
HostedWebSearchToolCurrent web information through Bing groundingNeeds a Bing connection, set as BING_CONNECTION_ID.
HostedMCPToolAn MCP server the service connects toConfigure approval_mode, allowed_tools, and headers. No context manager.
MCPStreamableHTTPToolAn MCP server your code connects toMust be opened with async with. Set auth headers through a custom http_client.
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.

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. 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.

Related

Last updated 2026-10-21