Advanced framework guide

Advanced Microsoft Agent Framework

Scale from per-run retrieval to multi-step agents that re-select tools every chat round, wire multi-agent workflows, and add governance middleware without leaving AF's native APIs.

← Basic Agent Framework usage

Per-call mode: re-select tools every chat round

Use query_strategy="per_call" when the agent chains multiple tool calls in one run and needs a different tool surface at each stage. The paired chat middleware is required — without it, per-call mode silently degrades to per-run and Gantry logs a one-time warning.

Per-call routing with middleware

from agent_framework import Agent
from agent_framework.openai import OpenAIChatClient
from agent_gantry.agent_framework import AgentFrameworkAdapter

provider = AgentFrameworkAdapter(gantry).context_provider(
    top_k=3,
    query_strategy="per_call",
)

agent = Agent(
    OpenAIChatClient(),
    "You are a multi-step assistant.",
    context_providers=[provider],
    middleware=[provider.as_chat_middleware()],  # required for per_call
)

# Or attach in one call:
# provider.attach_to(agent, trace=True)

Pin required and static tools

provider = AgentFrameworkAdapter(gantry).context_provider(
    top_k=5,
    required=["validate_input"],       # MissingRequiredToolError if absent
    always_include=["log_event"],      # warn + skip if absent
    static_tools=[some_af_native_tool],  # AF @tool callables outside Gantry
)

Multi-agent orchestration

Drop Gantry-equipped agents into SequentialBuilder, HandoffBuilder, or WorkflowBuilder. Each participant's run fires its own provider pipeline, so every agent gets tools tailored to its role and the current conversation state.

from agent_framework.orchestrations import SequentialBuilder
from agent_gantry import GantryContextProvider

triage = Agent(
    client,
    "You triage customer requests.",
    name="Triage",
    context_providers=[GantryContextProvider(gantry, top_k=2)],
)
resolver = Agent(
    client,
    "Resolve the issue using available tools.",
    name="Resolver",
    context_providers=[GantryContextProvider(gantry, top_k=3)],
)

workflow = SequentialBuilder(participants=[triage, resolver]).build()
result = await workflow.run("My flight was cancelled, please rebook.")

GantryToolBridge workflow patterns

For fan-out routing where each node receives its own semantically-selected tool slice, use GantryToolBridge.build_workflow(...) or the one-liner helpers build_handoff_workflow / build_sequential_workflow.

from agent_gantry.integrations.agent_framework_bridge import GantryToolBridge
from agent_gantry.integrations.agent_framework_middleware import (
    GantryApprovalMiddleware,
    GantryObservabilityMiddleware,
)

bridge = GantryToolBridge(gantry)
middleware = [
    GantryApprovalMiddleware(security_policy),
    GantryObservabilityMiddleware(gantry),
]

workflow_agent = await bridge.build_workflow(
    agent_specs=[
        dict(client=client, query="triage classify routing", name="Triage", limit=2, ...),
        dict(client=client, query="billing invoices", name="Billing", limit=2, middleware=middleware, ...),
    ],
    edges=[("Triage", "Billing", lambda ctx: "invoice" in str(ctx).lower())],
    workflow_name="CustomerServiceWorkflow",
)

Approval and observability middleware

Route destructive tools through AF approval mode and emit structured telemetry on every tool invocation:

from agent_gantry.core.security import SecurityPolicy
from agent_gantry.schema.tool import ToolCapability

@gantry.register(capabilities=[ToolCapability.DELETE_DATA])
def delete_user_account(user_id: str) -> str:
    """Delete a user account. Destructive; requires human approval."""
    return f"deleted:{user_id}"

policy = SecurityPolicy(require_confirmation=["delete_*", "refund_*"])
middleware = [
    AgentFrameworkAdapter(gantry).approval_middleware(policy),
    AgentFrameworkAdapter(gantry).observability_middleware(),
]

AF 1.6+ concurrent workflows

AF 1.6.0 enables ContextVar-based instrumentation by default. Sequential workflows are unaffected. If you run truly concurrent workflows via asyncio.gather() or TaskGroup, call disable_af_instrumentation() at startup or pass disable_af_instrumentation=True to GantryToolBridge.

from agent_gantry import disable_af_instrumentation

disable_af_instrumentation()  # before constructing concurrent workflow agents

Production patterns

Observability: use provider.attach_to(agent, trace=True) for built-in console trace middleware, or gantry.on_tool_call(...) event hooks.
Selection history: inspect provider.selections after multi-step runs to debug routing.
Score thresholds: start with score_threshold=0.0; use score_threshold="relative:0.8" for length-robust filtering on long prompts.
Evaluation: replay representative prompts with fixed tool snapshots before enabling dynamic per-call retrieval in production.