Ecosystem & interop

Make governed agents first-class citizens elsewhere: consume MCP tools, serve over A2A, publish as a Microsoft 365 / Foundry custom-engine agent, emit a full-run OpenTelemetry span tree, and wire human-in-the-loop approvals into the audit chain. Everything here is optional and local-first — protocol SDKs are extras, and the telemetry is a no-op unless OpenTelemetry is installed and configured.

MCP — consume Model Context Protocol tools

load_mcp_tools(session) turns an MCP server’s tools into governed Tools (the schema comes from the server). MCP is async, so use them with run.aio(...). Install the client with pip install "cendor-sdk[mcp]".

from mcp import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters
from cendor.sdk import Agent, run, load_mcp_tools

params = StdioServerParameters(command="my-mcp-server")
async with stdio_client(params) as (r, w), ClientSession(r, w) as session:
    await session.initialize()
    tools = await load_mcp_tools(session)          # each MCP tool -> a governed Tool
    agent = Agent(name="assistant", model="gpt-4o", tools=tools, instructions="Use the tools.")
    result = await run.aio(agent, "…")             # MCP calls flow through the bus/audit/budget

import { Agent, run, loadMcpTools } from '@cendor/sdk';

// `session` is your MCP client session — the duck-typed shape of @modelcontextprotocol/sdk's
// `Client` (camelCase `listTools()` / `callTool(name, args)`); a fake session tests it offline.
const tools = await loadMcpTools(session);       // each MCP tool -> a governed Tool
const agent = new Agent({ name: 'assistant', model: 'gpt-4o', tools, instructions: 'Use the tools.' });
const result = await run(agent, '…');            // MCP calls flow through the bus/audit/budget

The integration is duck-typed against a session with async list_tools() / call_tool(name, args), so a fake session tests it offline. load_mcp_resources(session) reads resources into {uri: contents}; load_mcp_prompts / get_mcp_prompt cover prompts.

A2A — serve an agent over the Agent-to-Agent protocol

from cendor.sdk import Agent, A2AServer, A2AClient

agent  = Agent(name="greeter", model="gpt-4o", instructions="Greet.")
server = A2AServer(agent)

# In-process (offline / embedded):
client = A2AClient(server)
print(client.card())               # the A2A agent card (name, skills, IO modes)
print(client.send("hi"))           # -> the agent's reply
full = client.send_full("hi")      # full A2A result incl. governance metadata:
                                   #   {"trace_id": ..., "cost_usd": ..., "agents": [...]}

# Over local HTTP (optional, opt-in — stdlib only):
from cendor.sdk.a2a import serve
httpd = serve(agent, host="127.0.0.1", port=8080)   # GET /.well-known/agent-card.json ; POST /
# httpd.serve_forever()  (run in a thread; httpd.shutdown() to stop)

import { Agent, A2AServer, A2AClient, serve } from '@cendor/sdk';

const agent  = new Agent({ name: 'greeter', model: 'gpt-4o', instructions: 'Greet.' });
const server = new A2AServer(agent);

// In-process (offline / embedded):
const client = new A2AClient(server);
console.log(client.card());              // the A2A agent card (name, skills, IO modes)
console.log(await client.send('hi'));    // -> the agent's reply
const full = await client.sendFull('hi'); // full A2A result incl. governance metadata:
                                         //   { metadata: { trace_id, cost_usd, agents } }

// Over local HTTP (optional, opt-in — node:http only):
const httpd = serve(agent, { host: '127.0.0.1', port: 8080 }); // GET /.well-known/agent-card.json ; POST /
// httpd.close() to stop

Note what rides along: the A2A reply carries the run’s trace_id and cost, so a consumer of your agent sees governed metadata, not just text.

Microsoft 365 / Foundry — publish as a custom-engine agent

FoundryAdapter speaks the Bot Framework Activity protocol — the surface a custom-engine agent exposes to Copilot / Teams / Azure AI Foundry:

from cendor.sdk import Agent, FoundryAdapter

adapter = FoundryAdapter(Agent(name="assistant", model="gpt-4o", instructions="Help."))

# In your web endpoint:
reply = adapter.on_activity(incoming_activity)   # -> outgoing message Activity, or None
# reply["channelData"]["cendor"] carries {"trace_id", "cost_usd", "agents"}
adapter.manifest()                                # minimal registration manifest

import { Agent, FoundryAdapter } from '@cendor/sdk';

const adapter = new FoundryAdapter(new Agent({ name: 'assistant', model: 'gpt-4o', instructions: 'Help.' }));

// In your web endpoint:
const incomingActivity = { type: 'message', text: 'hi', from: { id: 'user' } };
const reply = await adapter.onActivity(incomingActivity);  // -> outgoing message Activity, or null
// reply?.channelData.cendor carries { trace_id, cost_usd, agents }
adapter.manifest();                                        // minimal registration manifest

OpenTelemetry — a full-run `gen_ai` span tree

span_tree(result) emits a gen_ai.* span tree for a completed run, so the whole trajectory shows up in Foundry / Datadog / Jaeger: a root agent.run, a child per agent segment, and a grandchild per model call (chat {model}) and tool execution (execute_tool {name}). For live spans as the run progresses, wrap it in with live_spans():. Uses OpenTelemetry directly (extra [otel]) and is a no-op returning False if OTel isn’t installed.

from cendor.sdk import run
from cendor.sdk.otel import span_tree

result = run(agent, "…")
span_tree(result)     # spans exported to your configured OTel pipeline; mirrors result's tree

import { run, spanTree, liveSpans } from '@cendor/sdk';

const result = await run(agent, '…');
spanTree(result);     // spans exported to your configured OTel pipeline; mirrors result's tree

const live = liveSpans();   // or stream spans as the run progresses…
await run(agent, '…');
live.close();               // …and stop

Spans carry gen_ai.request.model, gen_ai.system, gen_ai.usage.input_tokens / output_tokens, gen_ai.usage.cost, and per-agent gen_ai.agent.name.

Human-in-the-loop — approvals in the audit chain

acttrace records that oversight happened; the pause/approve/resume mechanics are your app’s job. require_approval wraps a tool so each call consults an approver (the pause) and records the verdict via decision.human_oversight(...) on the same audit chain the run is correlated by:

from cendor.sdk import Agent, run, AuditLog
from cendor.sdk.hitl import require_approval

def approver(tool_name, args):
    return args["amount"] < 100, "auto-approved under $100"   # -> (approved, note)

refund = require_approval(issue_refund, approver=approver, reviewer="ops@bank")
agent  = Agent(name="support", model="gpt-4o", tools=[refund], instructions="Use tools.")

log = AuditLog(system="support", risk_tier="high", path="audit.jsonl")
run(agent, "Refund order 42 for $50", audit=log)
# audit.jsonl gains a human_oversight entry (approved/rejected), hash-chained & verify()-able.
# On rejection the real tool never runs and the model gets a denial to react to.

import { Agent, run, AuditLog, requireApproval } from '@cendor/sdk';

const approver = (toolName, args): [boolean, string] =>
  [args.amount < 100, 'auto-approved under $100'];   // -> [approved, note]

const refund = requireApproval(issueRefundTool, { approver, reviewer: 'ops@bank' });
const agent  = new Agent({ name: 'support', model: 'gpt-4o', tools: [refund],
                           instructions: 'Use tools.' });

const audit = new AuditLog('support', { riskTier: 'high', path: 'audit.jsonl' });
await run(agent, 'Refund order 42 for $50', { audit });
// audit.jsonl gains a human_oversight entry (approved/rejected), hash-chained & verify()-able.
// On rejection the real tool never runs and the model gets a denial to react to.

In production the approver is where you block on a reviewer — a prompt, a queue, a webhook — then return the decision to resume (or deny) the run.

Honest limits

MCP stdio transport is a local-process affair — on edge runtimes use HTTP/SSE transports.
A2A’s built-in HTTP server is stdlib-simple — fine for local and embedded use; put a real server in front of it for production traffic.
span_tree exports; it doesn’t collect. You still need an OTel pipeline (collector, backend) configured in your process.
require_approval is synchronous at the tool boundary — a long human pause holds the run open; for hours-long approvals, checkpoint the run and resume it (hardening).