Microsoft Agent Framework Overview

@tool(
    name="lookup_expense_policy",
    description="Look up the reimbursement policy for an expense category.",
)
def lookup_expense_policy(category: str) -> str:
    """Return the reimbursement policy for a category.
 
    Args:
        category: Expense category, e.g. 'meals' or 'travel'.
 
    Returns:
        A short policy summary string.
    """
    policy = _POLICIES.get(category.strip().lower())
    if policy is None:
        return f"No specific policy for '{category}'; the general limit is $200 with receipts required."
    return (
        f"Policy for {category}: limit ${policy['limit_usd']} per expense, "
        f"receipts required over ${policy['receipt_required_over_usd']}. {policy['notes']}"
    )

@tool(
    name="submit_expense",
    description="Submit an expense report entry for reimbursement. Requires human approval.",
    approval_mode="always_require",
)
def submit_expense(expense: Expense) -> str:
    """Submit the expense for reimbursement once a human approves it.
 
    Args:
        expense: The complete expense entry (vendor, category, amount_usd, memo).
 
    Returns:
        A confirmation string with the recorded amount.
    """
    # On the approval-resume path the framework replays the stored tool-call
    # arguments as plain dicts rather than re-validating through pydantic —
    # normalize before reading attributes.
    entry = expense if isinstance(expense, Expense) else Expense.model_validate(expense)
    return (
        f"Expense recorded: ${entry.amount_usd:.2f} to {entry.vendor} "
        f"({entry.category}) — queued for reimbursement."
    )

policy_researcher = Agent(
    name="policy_researcher",
    instructions=(
        "You are an expense-policy researcher. Given an expense category and "
        "amount, summarize the applicable policy rules in 3 short bullets."
    ),
    client=build_chat_client(),
)
 
 
@tool(
    name="research_policy",
    description="Delegate policy research for this expense to a specialist.",
)
async def research_policy(category: str, amount: float) -> str:
    """Delegate policy research for this expense to a specialist.
 
    Streams the ``policy_researcher`` specialist and mirrors each text delta
    onto the AG-UI wire as attributed SUBAGENT_* / TEXT_MESSAGE_* events via
    src/subagent_emitter.py (no-ops outside the wrapped run).
 
    Args:
        category: Expense category, e.g. 'meals' or 'travel'.
        amount: Expense amount in USD.
 
    Returns:
        The specialist's complete policy summary.
    """
    # Deterministically recorded by the run wrapper's pump before this body
    # runs (the bridge streams TOOL_CALL_START/ARGS/END first); None when
    # invoked outside a wrapped run.
    tid = subagent_emitter.current_tool_call_id("research_policy")
    subagent_emitter.delegation_started(tid, policy_researcher.name)
    parts: list[str] = []
    try:
        prompt = (
            f"Expense category: {category}. Amount: ${amount:.2f}. "
            "Summarize the applicable policy rules."
        )
        async for update in policy_researcher.run(prompt, stream=True):
            text = update.text
            if text:
                parts.append(text)
                subagent_emitter.delegation_delta(tid, text)
    except Exception as exc:
        subagent_emitter.delegation_error(tid, str(exc))
        raise
    subagent_emitter.delegation_finished(tid)
    return "".join(parts)

def build_chat_client() -> OpenAIChatCompletionClient:
    """Azure OpenAI by default; plain OpenAI when Azure env is absent.
 
    Passing `azure_endpoint` explicitly is the constructor's strongest Azure
    signal — it wins even when OPENAI_API_KEY is also set, which makes Azure
    the default whenever it is configured. Key, deployment (model), and API
    version resolve from AZURE_OPENAI_* env vars.
    """
    azure_endpoint = os.environ.get("AZURE_OPENAI_ENDPOINT")
    if azure_endpoint:
        return OpenAIChatCompletionClient(
            model=os.environ.get("AZURE_OPENAI_MODEL"),
            azure_endpoint=azure_endpoint,
        )
    # Passing api_key explicitly forces OpenAI routing even when no
    # OPENAI_API_KEY env var is set; without it the constructor falls back to
    # Azure env resolution and raises at import time. A placeholder key fails
    # properly at request time (401) instead of at module import.
    return OpenAIChatCompletionClient(
        model=os.environ.get("OPENAI_CHAT_MODEL", "gpt-4o-mini"),
        api_key=os.environ.get("OPENAI_API_KEY", "unset-openai-api-key"),
    )

agent = AgentFrameworkAgent(
    agent=Agent(
        name="expense_approval_copilot",
        instructions=_INSTRUCTIONS,
        client=build_chat_client(),
        tools=[lookup_expense_policy, research_policy, submit_expense],
    ),
    name="ExpenseApprovalCopilot",
    description="Files expense reports with policy lookup, shared state, and human approval.",
    state_schema={
        "expense": {"type": "object", "description": "The expense entry being drafted."},
    },
    predict_state_config={
        "expense": {"tool": "submit_expense", "tool_argument": "expense"},
    },
    require_confirmation=False,
)

from fastapi import FastAPI
from agent_framework_ag_ui import add_agent_framework_fastapi_endpoint
 
from .agent import agent
from .subagent_emitter import wrap_agent_run
 
app = FastAPI(title="cockpit-runtimes-microsoft-agent-framework")
# The wrapper is the SUBAGENT_* injection seam: the endpoint consumes
# protocol_runner.run, and wrap_agent_run merges the delegation tool's
# enqueued child events into that stream (src/subagent_emitter.py).
wrapped_agent = wrap_agent_run(agent)
add_agent_framework_fastapi_endpoint(app, wrapped_agent, path="/agent")
 
 
@app.get("/ok")
def ok() -> dict:
    return {"ok": True}

import { injectCockpitRuntimeConnection } from '@threadplane/cockpit-telemetry';
import { ApplicationConfig } from '@angular/core';
import { provideAgent } from '@threadplane/ag-ui';
 
export const appConfig: ApplicationConfig = {
  providers: [
    provideAgent(() => {
      const connection = injectCockpitRuntimeConnection();
      if (connection.adapter !== 'ag-ui') {
        throw new Error('incompatible runtime');
      }
      return {
        url: connection.url,
      };
    }),
  ],
};

/** Shared state streamed from the backend (STATE_SNAPSHOT / STATE_DELTA). */
protected readonly expense = computed(() => {
  // Example apps compile lib source with strict:false — cast at the read site.
  const state = this.agent.state() as { expense?: ExpenseDraft } | undefined;
  const e = state?.expense;
  return e && e.vendor !== undefined ? e : undefined;
});

/**
 * The pending approval request from the protocol-standard interrupt
 * outcome. The reducer stores it as `{ interrupts: [...], runId }`; each
 * entry's `metadata.agent_framework.function_call` carries the tool name
 * and parsed arguments.
 */
private readonly approvalCall = computed(() => {
  const value = this.agent.interrupt?.()?.value as { interrupts?: unknown[] } | undefined;
  const first = value?.interrupts?.[0] as
    | { metadata?: { agent_framework?: { function_call?: { name?: string; arguments?: unknown } } } }
    | undefined;
  return first?.metadata?.agent_framework?.function_call;
});
 
protected readonly approvalToolName = computed(() => this.approvalCall()?.name ?? 'a tool call');
 
protected readonly approvalExpense = computed(() => {
  const args = this.approvalCall()?.arguments as { expense?: ExpenseDraft } | undefined;
  return args?.expense;
});
 
protected send(text: string): void {
  void this.agent.submit({ message: text });
}
 
protected onAction(action: ChatApprovalAction): void {
  if (action === 'approve') {
    void this.agent.submit({ resume: { approved: true } });
  } else if (action === 'cancel') {
    void this.agent.submit({ resume: { approved: false } });
  }
}