1"""Reference hosted agent: GitHub Copilot SDK + Foundry toolbox (tools and skills over MCP).
2
3Reference only: copilot-sdk-foundry.md is the recipe and wins on any conflict. Written against
4github-copilot-sdk 1.0.14 and re-checked on 1.0.20; inspect the installed APIs first. Call `run_turn` from the Responses
5handler that `azd ai agent init` scaffolds: create one credential and one client at startup and
6close them at shutdown, pass the request's `x-agent-foundry-call-id` from the adapter's verified
7request-context accessor, and stream the returned tool calls as output items before the message
8(foundry-hosted-agent.md). Return a ToolboxError that has a consent_url to the caller as the
9answer: the first use of an OAuth connection needs one sign-in per user.
10
11    pip install "github-copilot-sdk[telemetry]" azure-identity
12
13Environment (the agent's `env` in azure.yaml; FOUNDRY_* and AGENT_* are platform-reserved):
14

copilot-sdk-with-toolbox.py on GitHub

TOOLBOX_ENDPOINT Consumer URL <project>/toolboxes/<name>/mcp?api-version=v1, mapped

from TOOLBOX_<NAME>_MCP_ENDPOINT; it serves the default version.

MODEL_API_BASE_URL The discovered .../openai/v1/ base and its deployment. Required

AZURE_AI_MODEL_DEPLOYMENT_NAME when hosted; locally, unset means the Copilot default model.

COPILOT_OTLP_ENDPOINT Optional OTLP HTTP receiver for runtime spans.

Skills attached to the toolbox are MCP resources (skill://<name>/SKILL.md) that follow each

skill's default version, so publishing a skill needs no redeploy. For multi-file skills, or

skills outside a toolbox, use the Skills API download in microsoft-foundry. Verify that served

bodies keep their frontmatter; load_skills rebuilds it when missing.

"""

from __future__ import annotations

import asyncio

import json

import logging

import os

import re

import tempfile

from collections.abc import Callable

from dataclasses import dataclass, field

from pathlib import Path

from typing import Any

import httpx

from azure.core.credentials_async import AsyncTokenCredential

from azure.identity.aio import DefaultAzureCredential

from copilot import CopilotClient, SessionEventType, Tool, ToolInvocation, ToolResult

from copilot.generated.rpc import PermissionDecisionApproveOnce, PermissionDecisionReject

from copilot.generated.session_events import PermissionRequestCustomTool, PermissionRequestRead

from copilot.session import ProviderConfig, ProviderTokenArgs

from opentelemetry import trace

from opentelemetry.trace import Status, StatusCode

log = logging.getLogger("agent")

tracer = trace.get_tracer(__name__)

SCOPE = "https://ai.azure.com/.default" # model and toolbox audience; cognitiveservices gets 401

TOOLBOX_ENDPOINT = os.environ["TOOLBOX_ENDPOINT"]

MODEL_BASE_URL = os.getenv("MODEL_API_BASE_URL")

MODEL_DEPLOYMENT = os.getenv("AZURE_AI_MODEL_DEPLOYMENT_NAME")

if bool(MODEL_BASE_URL) != bool(MODEL_DEPLOYMENT):

# One without the other would silently fall back to the Copilot default model.

raise RuntimeError("Set both MODEL_API_BASE_URL and AZURE_AI_MODEL_DEPLOYMENT_NAME, or neither.")

WRITABLE = Path(tempfile.gettempdir()) # hosted containers are read-only except the temp dir

SKILL_URI = re.compile(r"skill://([A-Za-z0-9_-]+)/SKILL\.md")

TURN_TIMEOUT = 300.0

@dataclass

class ToolCall:

"""One toolbox call, emitted as function_call and function_call_output items."""

call_id: str

name: str

arguments: str

output: str

@dataclass

class Turn:

text: str

tool_calls: list[ToolCall] = field(default_factory=list)

class ToolboxError(RuntimeError):

"""A toolbox JSON-RPC error; `consent_url` is set when a connection needs user consent."""

def __init__(self, message: str, consent_url: str | None = None) -> None:

super().__init__(message)

self.consent_url = consent_url

def _toolbox_error(error: dict[str, Any]) -> ToolboxError:

message = str(error.get("message", ""))

if error.get("code") == -32006 and "{" in message:

try: # -32006 wraps per-source errors as JSON after a human-readable prefix

for source in json.loads(message[message.index("{") :]).get("errors", []):

nested = source.get("error") or {}

if nested.get("code") == "CONSENT_REQUIRED":

url = nested.get("message")

name = source.get("name", "A toolbox connection")

return ToolboxError(f"{name} needs consent: open {url}, sign in, then retry.", url)

except (ValueError, AttributeError):

pass

return ToolboxError(f"Toolbox error {error.get('code')}: {message or 'unknown'}")

class _Bearer(httpx.Auth):

"""Adds a token to every request; Azure Identity caches and renews it."""

def __init__(self, credential: AsyncTokenCredential) -> None:

self._credential = credential

async def async_auth_flow(self, request: httpx.Request):

request.headers["Authorization"] = f"Bearer {(await self._credential.get_token(SCOPE)).token}"

yield request

class McpBridge:

"""Request-scoped JSON-RPC client for the toolbox MCP endpoint; initialize is optional."""

def __init__(self, credential: AsyncTokenCredential, call_id: str | None) -> None:

headers = {"Foundry-Features": "Toolboxes=V1Preview"}

if call_id: # forward the caller context unchanged; never fabricate one

headers["x-agent-foundry-call-id"] = call_id

self._http = httpx.AsyncClient(auth=_Bearer(credential), headers=headers, timeout=60.0)

self._next_id = 0

async def __aenter__(self) -> McpBridge:

return self

async def __aexit__(self, *_exc: object) -> None:

await self._http.aclose()

async def rpc(self, method: str, params: dict[str, Any], target: str = "") -> dict[str, Any]:

"""One JSON-RPC call in an MCP client span that records no arguments or results."""

self._next_id += 1

payload = {"jsonrpc": "2.0", "id": self._next_id, "method": method, "params": params}

name = f"{method} {target}".strip()

with tracer.start_as_current_span(

name, record_exception=False, set_status_on_exception=False

) as span:

span.set_attribute("mcp.method.name", method)

if method == "tools/call":

span.set_attribute("gen_ai.tool.name", target)

try:

response = await self._http.post(TOOLBOX_ENDPOINT, json=payload)

response.raise_for_status()

body = response.json()

if "error" in body:

raise _toolbox_error(body["error"])

except Exception as exc:

span.set_status(Status(StatusCode.ERROR))

span.set_attribute("error.type", type(exc).__name__)

raise

return body.get("result") or {}

def _call_handler(toolbox: McpBridge, mcp_name: str, calls: list[ToolCall]):

async def handler(invocation: ToolInvocation) -> ToolResult:

arguments = invocation.arguments if isinstance(invocation.arguments, dict) else {}

try:

result = await toolbox.rpc("tools/call", {"name": mcp_name, "arguments": arguments}, mcp_name)

except Exception as exc: # report to the model; it can retry or explain

output, outcome = str(exc), ToolResult(result_type="failure", error=str(exc))

else:

texts = [c.get("text", "") for c in result.get("content", []) if c.get("type") == "text"]

output = "\n".join(t for t in texts if t) or json.dumps(result)

failed = bool(result.get("isError"))

outcome = ToolResult(

text_result_for_llm=output,

result_type="failure" if failed else "success",

error=output if failed else None,

)

calls.append(ToolCall(invocation.tool_call_id, mcp_name, json.dumps(arguments), output))

return outcome

return handler

def toolbox_tools(toolbox: McpBridge, listed: list[dict[str, Any]], calls: list[ToolCall]) -> list[Tool]:

"""Wrap each tools/list entry as an SDK tool; MCP-sourced names look like label___tool."""

tools: dict[str, Tool] = {}

sources: dict[str, str] = {}

for spec in listed:

mcp_name = spec["name"]

name = re.sub(r"[.-]", "_", mcp_name) # SDK tool names reject dots and hyphens

if name in sources:

# Dropping one would silently change the agent's capabilities; rename a tool instead.

raise ValueError(f"Toolbox tools {sources[name]!r} and {mcp_name!r} both map to {name!r}.")

sources[name] = mcp_name

tools[name] = Tool(

name=name,

description=spec.get("description") or mcp_name,

parameters=spec.get("inputSchema") or {"type": "object", "properties": {}},

handler=_call_handler(toolbox, mcp_name, calls),

)

return list(tools.values())

async def load_skills(toolbox: McpBridge, skills_dir: Path) -> list[str]:

"""Write each toolbox skill to <skills_dir>/<name>/SKILL.md for the SDK skill loader."""

names: list[str] = []

for resource in (await toolbox.rpc("resources/list", {})).get("resources", []):

match = SKILL_URI.fullmatch(resource.get("uri", ""))

if not match:

continue # skill://index.json, archives and other resources

name = match.group(1)

read = await toolbox.rpc("resources/read", {"uri": resource["uri"]}, name)

body = "".join(c.get("text", "") for c in read.get("contents", []))

if not body.lstrip().startswith("---"):

description = json.dumps(resource.get("description") or name)

body = f"---\nname: {name}\ndescription: {description}\n---\n\n{body}"

(skills_dir / name).mkdir(parents=True, exist_ok=True)

(skills_dir / name / "SKILL.md").write_text(body, encoding="utf-8")

names.append(name)

return names

def model_provider(credential: AsyncTokenCredential) -> ProviderConfig:

async def refresh_token(_request: ProviderTokenArgs) -> str:

return (await credential.get_token(SCOPE)).token

return {

"type": "openai",

"base_url": MODEL_BASE_URL,

"wire_api": "responses",

"bearer_token_provider": refresh_token,

}

def least_privilege(tool_names: set[str], skills_dir: Path) -> Callable[[Any, Any], Any]:

"""Approve this turn's toolbox tools and reads of its skills; reject everything else."""

def decide(request: Any, _invocation: Any) -> Any:

if isinstance(request, PermissionRequestCustomTool) and request.tool_name in tool_names:

return PermissionDecisionApproveOnce()

if isinstance(request, PermissionRequestRead):

if Path(request.resolved_path or request.path).resolve().is_relative_to(skills_dir):

return PermissionDecisionApproveOnce()

return PermissionDecisionReject(feedback="Not permitted for this agent.")

return decide

def log_event(event: Any) -> None:

"""Log safe identifiers only: never prompts, arguments, results or caller context."""

data = event.data

if event.type == SessionEventType.SESSION_ERROR:

log.error(

"session error type=%s code=%s status=%s request=%s",

data.error_type, data.error_code, data.status_code, data.service_request_id,

)

elif event.type == SessionEventType.TOOL_EXECUTION_START:

log.info("tool start %s id=%s", data.tool_name, data.tool_call_id)

elif event.type == SessionEventType.TOOL_EXECUTION_COMPLETE:

log.info("tool done id=%s success=%s", data.tool_call_id, data.success)

def new_client() -> CopilotClient:

"""One Copilot runtime per process."""

options: dict[str, Any] = {}

if not os.access(Path.home(), os.W_OK): # session state defaults to ~/.copilot

options["base_directory"] = str(WRITABLE / "copilot")

if endpoint := os.getenv("COPILOT_OTLP_ENDPOINT"):

options["telemetry"] = {

"otlp_endpoint": endpoint,

"otlp_protocol": "http/protobuf",

"exporter_type": "otlp-http",

"capture_content": False,

}

return CopilotClient(**options)

async def run_turn(

client: CopilotClient, credential: AsyncTokenCredential, prompt: str, call_id: str | None = None

) -> Turn:

"""Run one request with its own toolbox connection, skills directory and session."""

calls: list[ToolCall] = []

headers = {"x-agent-foundry-call-id": call_id} if call_id else None

options: dict[str, Any] = {}

if MODEL_BASE_URL:

options = {"provider": model_provider(credential), "model": MODEL_DEPLOYMENT}

async with McpBridge(credential, call_id) as toolbox:

tools = toolbox_tools(toolbox, (await toolbox.rpc("tools/list", {})).get("tools", []), calls)

with tempfile.TemporaryDirectory(prefix="turn-", dir=WRITABLE) as workdir:

skills_dir = Path(workdir).resolve() / "skills"

await load_skills(toolbox, skills_dir)

async with await client.create_session(

**options,

tools=tools,

skill_directories=[str(skills_dir)],

working_directory=workdir,

on_permission_request=least_privilege({tool.name for tool in tools}, skills_dir),

on_event=log_event,

) as session:

try:

reply = await session.send_and_wait(

prompt, request_headers=headers, timeout=TURN_TIMEOUT

)

except TimeoutError:

await session.abort() # the timeout alone leaves the turn running

raise

if reply is None or not reply.data.content:

raise RuntimeError("The session ended without an assistant message.")

return Turn(reply.data.content, calls)

async def main() -> None:

"""Local smoke run: `az login`, set TOOLBOX_ENDPOINT, optionally the model pair."""

logging.basicConfig(level=logging.INFO)

async with DefaultAzureCredential() as credential, new_client() as client:

try:

turn = await run_turn(client, credential, "Which tools and skills can you use? Be brief.")

except ToolboxError as exc:

if exc.consent_url is None:

raise

print(exc)

return

for call in turn.tool_calls:

print(f"tool {call.name}: {len(call.output)} chars")

print(turn.text)

if __name__ == "__main__":

asyncio.run(main())