Build an Agent

This guide builds an autonomous agent on the pattern of Revenue Sentinel (brutor-revenue-sentinel-agent), a real sample agent that scans HubSpot for revenue risks via MCP, reasons with an LLM, and posts alerts to Slack — with every step routed through the gateway.

The key idea: the agent contains zero governance code. PII redaction, tool filters, budgets, human approvals, and the audit trail all come from the gateway, because both the LLM calls and the tool calls point at it:

Without the gateway                 With the gateway
───────────────────                 ────────────────
Agent → OpenAI (raw)                Agent → :8100/v1/proxy/llm  → OpenAI
Agent → HubSpot API (full PII)      Agent → :8100/v1/proxy/mcp/{hubspot} → HubSpot (PII redacted)
Agent → Slack (anything)           Agent → :8100/v1/proxy/mcp/{slack}   → Slack (writes logged, filtered)
No audit trail                      Every step in Mission Control

Same agent. Same data. Now it’s governed.

Prerequisites

• The gateway running (see Quickstart)
• An AI System for the agent to run as — a resource group with type AI System (create one; the full production setup is in Your AI System → Deploy)
• A model bound to it (Resources → AI Models → Configuration → Edit → Access Control) and one or more MCP servers bound under Resource Groups → your system → Resources → MCP Servers (see Register your own MCP server)
• An API key minted on that system’s API Keys tab (sk_brutor_api_...)

pip install langgraph openai "mcp[cli]" httpx python-dotenv

1. Configuration — every URL points at the gateway

config.py

import os
from dotenv import load_dotenv

load_dotenv()

GATEWAY = os.getenv("GATEWAY_PROXY_URL", "http://localhost:8100/v1/proxy")
GATEWAY_API_KEY = os.environ["GATEWAY_API_KEY"]          # sk_brutor_api_...

LLM_URL = f"{GATEWAY}/llm"                               # OpenAI-compatible
# Server IDs come from Admin UI → Resources → MCP Servers (the "endpoint" link)
HUBSPOT_MCP_URL = f"{GATEWAY}/mcp/{os.environ['HUBSPOT_SERVER_ID']}"
SLACK_MCP_URL   = f"{GATEWAY}/mcp/{os.environ['SLACK_SERVER_ID']}"

CHAT_MODEL = os.getenv("CHAT_MODEL", "gpt-4o")   # any model bound to your AI System

2. The LLM client — one base-URL change

llm.py

from openai import AsyncOpenAI
from config import GATEWAY_API_KEY, LLM_URL

client = AsyncOpenAI(
    api_key=GATEWAY_API_KEY,     # the Brutor API key IS the OpenAI api_key
    base_url=LLM_URL,            # http://localhost:8100/v1/proxy/llm
)

That’s the entire integration for the reasoning side. Chat completions, streaming, and legacy completions all work — see Call LLMs through the gateway.

3. MCP tools through the gateway

The gateway speaks MCP streamable HTTP at /v1/proxy/mcp/{server_id}, so the standard MCP Python SDK connects to it like any other MCP server — you just add the API key header. This is the exact pattern from Revenue Sentinel’s mcp_client.py:

mcp_client.py

import httpx
from contextlib import AsyncExitStack
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from config import GATEWAY_API_KEY

class MCPClient:
    """A persistent MCP session to one server, via the gateway."""

    def __init__(self, server_url: str, name: str):
        self.server_url = server_url
        self.name = name
        self._exit_stack = AsyncExitStack()
        self._session: ClientSession | None = None

    async def connect(self) -> ClientSession:
        if self._session:
            return self._session
        headers = {"Authorization": f"Bearer {GATEWAY_API_KEY}"}
        http_client = await self._exit_stack.enter_async_context(
            httpx.AsyncClient(headers=headers, timeout=None)
        )
        read, write, _ = await self._exit_stack.enter_async_context(
            streamable_http_client(self.server_url, http_client=http_client)
        )
        self._session = await self._exit_stack.enter_async_context(
            ClientSession(read, write)
        )
        await self._session.initialize()
        return self._session

    async def get_tools_for_openai(self) -> list[dict]:
        """MCP tool defs → OpenAI tool-calling format, name-prefixed per server."""
        session = await self.connect()
        result = await session.list_tools()
        return [
            {
                "type": "function",
                "function": {
                    "name": f"{self.name}__{tool.name}",
                    "description": f"[{self.name.upper()}] {tool.description or ''}",
                    "parameters": tool.inputSchema,
                },
            }
            for tool in result.tools
        ]

    async def call_tool(self, name: str, arguments: dict):
        session = await self.connect()
        result = await session.call_tool(name, arguments=arguments)
        return result.content

    async def aclose(self):
        await self._exit_stack.aclose()
        self._session = None

Prefixing tool names with the server name (hubspot__get_contacts, slack__post_message) avoids collisions when the model sees tools from several servers at once; a small registry maps the prefixed name back to (server, original_name) at dispatch time (see MultiMCPClient in the sample repo).

What the gateway does on each tools/call

Tool filters (is this tool enabled for your group?), input guardrails on arguments (mcp_input surface), argument policies (e.g. deny SQL writes), approval gates, output guardrails on results (mcp_output — where PII redaction happens), and a full audit log row. None of that appears in the code above.

4. The LangGraph loop

The classic agentic loop: the model reasons, requests tools, you execute them via MCP, feed results back, repeat until it stops asking.

graph.py

import json
import operator
from typing import Annotated, List, Literal, TypedDict

from langgraph.graph import StateGraph, START, END
from llm import client
from config import CHAT_MODEL, HUBSPOT_MCP_URL, SLACK_MCP_URL
from mcp_client import MCPClient

hubspot = MCPClient(HUBSPOT_MCP_URL, "hubspot")
slack = MCPClient(SLACK_MCP_URL, "slack")
CLIENTS = {"hubspot": hubspot, "slack": slack}

SYSTEM_PROMPT = """You are Revenue Sentinel, an autonomous sales intelligence agent.
Phase 1: gather deals, contacts, and recent activity from HubSpot (read-only).
Phase 2: analyse for churn risk, stalled pipeline, and unassigned high-value deals.
Phase 3: post severity-graded alerts to the right Slack channels.
Include relevant details in alerts — the gateway governance layer handles
redaction; that is not your job. Only flag what the data shows."""

class AgentState(TypedDict):
    messages: Annotated[List[dict], operator.add]
    available_tools: List[dict]

async def load_tools(state: AgentState):
    tools = []
    for c in CLIENTS.values():
        tools.extend(await c.get_tools_for_openai())
    return {"available_tools": tools}

async def agent(state: AgentState):
    response = await client.chat.completions.create(
        model=CHAT_MODEL,
        messages=[{"role": "system", "content": SYSTEM_PROMPT}, *state["messages"]],
        tools=state["available_tools"],
        tool_choice="auto",
    )
    return {"messages": [response.choices[0].message.model_dump(exclude_none=True)]}

async def execute_tools(state: AgentState):
    tool_messages = []
    for tc in state["messages"][-1].get("tool_calls", []):
        server_name, tool_name = tc["function"]["name"].split("__", 1)
        args = json.loads(tc["function"]["arguments"])
        try:
            content = await CLIENTS[server_name].call_tool(tool_name, args)
            result = "\n".join(p.text for p in content if hasattr(p, "text"))
        except Exception as e:
            result = f"Tool error ({tool_name}): {e}"
        tool_messages.append(
            {"role": "tool", "tool_call_id": tc["id"], "content": result}
        )
    return {"messages": tool_messages}

def should_continue(state: AgentState) -> Literal["execute_tools", "done"]:
    return "execute_tools" if state["messages"][-1].get("tool_calls") else "done"

workflow = StateGraph(AgentState)
workflow.add_node("load_tools", load_tools)
workflow.add_node("agent", agent)
workflow.add_node("execute_tools", execute_tools)
workflow.add_edge(START, "load_tools")
workflow.add_edge("load_tools", "agent")
workflow.add_conditional_edges("agent", should_continue,
                               {"execute_tools": "execute_tools", "done": END})
workflow.add_edge("execute_tools", "agent")
graph_app = workflow.compile()

Run it:

main.py

import asyncio
from graph import graph_app

async def main():
    await graph_app.ainvoke({
        "messages": [{"role": "user", "content": "Run the revenue scan."}],
        "available_tools": [],
    })

asyncio.run(main())

No LangGraph? The same pattern works as a plain while loop: call chat.completions.create with tools, execute any tool_calls via MCP, append the role: "tool" results, repeat.

5. Handling approval_required

Operators can mark tools as requires approval on a resource group (e.g. every Slack write). When your agent calls such a tool without prior approval, the gateway does not forward the call — it creates an approval request and returns HTTP 202 (for API-key callers) with a plain JSON body instead of a JSON-RPC result:

{
  "approval_required": true,
  "approval_request_id": "appr-01JX2M...",
  "capability_type": "tool",
  "capability_name": "post_message",
  "server_name": "slack",
  "surface": "mcp",
  "timeout_seconds": 300,
  "expires_at": "2026-07-02T12:34:56Z",
  "poll_url": "/v1/portal/approvals/appr-01JX2M.../poll",
  "message": "Tool 'post_message' requires approval. Submit with X-Approval-Token header after approval."
}

A group member sees the request in the portal’s approval inbox (or a notification) and approves or rejects it. Your agent’s flow:

1. Detect the 202. Because the body is not a JSON-RPC envelope, the MCP SDK surfaces it as a transport-level error. For tools that may be approval-gated, issue tools/call as a plain JSON-RPC POST so you can inspect the response body directly (below).

2. Poll the approval. GET /v1/portal/approvals/{id}/poll accepts your API key (agents don’t need a portal JWT for this one endpoint). While pending it returns {"id": "...", "status": "pending", "expires_at": "..."}; once a human approves, the response includes the one-time token:

   {"id": "appr-01JX2M...", "status": "approved", "approval_token": "apt_9f2c..."}

3. Retry with X-Approval-Token. Re-send the same tool call with the token in the header. The token is one-time — it authorises exactly one submission of that capability.

approval.py

import asyncio, httpx
from config import GATEWAY_API_KEY

GATEWAY_ROOT = "http://localhost:8100"

async def call_tool_with_approval(
    server_url: str, session_id: str, tool: str, arguments: dict,
    poll_interval: float = 3.0, timeout: float = 300.0,
) -> dict:
    """tools/call over raw JSON-RPC with approval_required handling."""
    headers = {
        "Authorization": f"Bearer {GATEWAY_API_KEY}",
        "Content-Type": "application/json",
        "Accept": "application/json, text/event-stream",
        "Mcp-Session-Id": session_id,
    }
    rpc = {"jsonrpc": "2.0", "id": 1, "method": "tools/call",
           "params": {"name": tool, "arguments": arguments}}

    async with httpx.AsyncClient(timeout=60) as http:
        resp = await http.post(server_url, json=rpc, headers=headers)

        if resp.status_code == 202:                      # approval gate
            info = resp.json()
            approval_id = info["approval_request_id"]
            print(f"⏸  '{tool}' needs approval — waiting ({info['message']})")

            deadline = asyncio.get_event_loop().time() + timeout
            token = None
            while asyncio.get_event_loop().time() < deadline:
                poll = await http.get(
                    f"{GATEWAY_ROOT}/v1/portal/approvals/{approval_id}/poll",
                    headers={"Authorization": f"Bearer {GATEWAY_API_KEY}"},
                )
                status = poll.json()
                if status["status"] == "approved":
                    token = status["approval_token"]
                    break
                if status["status"] in ("rejected", "expired"):
                    raise PermissionError(f"'{tool}' {status['status']} by operator")
                await asyncio.sleep(poll_interval)
            if token is None:
                raise TimeoutError(f"approval for '{tool}' timed out")

            # One-time token → retry the identical call
            resp = await http.post(server_url, json=rpc,
                                   headers={**headers, "X-Approval-Token": token})

        resp.raise_for_status()
        return resp.json()

(The Mcp-Session-Id comes from the initialize handshake — streamable-HTTP MCP is stateful; see the smoke test in Register your own MCP server.)

Feed the outcome back to the model as a normal role: "tool" message either way — “posted successfully” after approval, or “rejected by operator” — so it can adapt its plan.

Interactive callers get a 428 instead

Portal users (JWT-authenticated humans) receive 428 Precondition Required rather than 202, because a person is present to act on it synchronously. Agents and API-key callers always get 202 + poll.

6. Run correlation — one header, one run per task

Your agent drives its own loop, which means the gateway sees each step as an unrelated request: one task would land in the ledger as five separate one-step “runs”, and every lifecycle signal built on runs — cost per completed task, trajectory length, drift baselines, replay suites — would be computed over fragments.

The fix is one header on every call of a task, plus one on the last:

run_context.py

import uuid

def new_run_headers() -> dict:
    """Call once per task; send on EVERY gateway call the task makes."""
    return {"x-brutor-run-id": f"task-{uuid.uuid4()}"}

# ...inside your loop, on the FINAL call of the task:
headers = {**run_headers, "X-Brutor-Run-End": "completed", "X-Brutor-Run-Outcome": "resolved"}

Three things worth knowing about how this behaves:

• You cannot join someone else’s run. The gateway derives the real run root from your id keyed by the tenant secret and bound to your credential — the same id from a different API key resolves to a different run. There is nothing to guess and nothing to forge.
• The ledger says how actions were linked. Your actions record trace_continuity: client_asserted — complete coverage, client-word linkage — distinct from the verified a gateway-signed delegation chain earns (which is what your escalations to other agents get automatically). The run’s integrity column keeps the two apart rather than rounding both up to “intact”.
• Close your runs. X-Brutor-Run-End: completed (or errored, exhausted, cancelled) closes the run immediately with an honest outcome. A run nobody closes is swept as abandoned after its idle window — indistinguishable from a crash — and exhausted in particular can only come from you: the gateway cannot see your framework’s iteration cap, and it refuses to guess. X-Brutor-Run-Outcome is the other thing only you know — whether the task was resolved, escalated, handed off or abandoned by the user. The ledger records it as your claim, with its own coverage, never folded into the terminal state it derived.

Optional: name your phases

If your agent has distinct stages — a LangGraph node, a milestone on its own plan — you can name them, and the ledger will attribute cost and behaviour per phase:

X-Brutor-Step-Id:   research          # stable for the whole phase
X-Brutor-Step-Name: Gather claim data # optional label, for display

This is worth doing when you want to answer “which phase is burning the budget” or “which tools ran during settlement” — the Trajectory Explorer is built on it.

Why we never guess your phases

A turn boundary the gateway can work out for itself: a model call opens one. A step boundary it cannot — nothing in the traffic distinguishes “still researching” from “now drafting”, because that decision lives in your orchestrator. So if you send nothing, the run reads unphased rather than being credited with a single phase it never declared. Absence of your assertion is recorded as absence, not rounded up.

If your agent is invoked as a delegated peer (another agent reached you through the gateway’s A2A proxy), echo the x-brutor-delegation-root, -parent and -depth headers you received instead — a signed chain always outranks an assertion, and an action that belongs to someone else’s delegation must not re-root itself.

If your stack already runs OpenTelemetry, send the standard traceparent header as well: the gateway records its trace and span ids on every audit row the call produces, so you can jump from a span in your tracing tool to the governed record. The record also names the model whose completion decided each tool call, and the hash of the tool definition that was in force — no header needed for either.

What you get for free

Run the agent, then open AI Estate → AI Systems → your system:

• Runs — one run per task, with an honest terminal state, its cost, its tools and its chain integrity; Details on a row shows everything the ledger holds about it, and Trajectory replays it turn by turn, phase by phase if you named them
• Signals — liveness against the expectation you set, cost per completed task, trajectory length, errors and outcomes, all learned into a baseline that drift detection later compares against
• Lifecycle and Contract — the stage the system is in, and the hash-pinned snapshot of exactly what it was allowed to use when each run executed

And in Mission Control: the system’s card on the Overview (spend, cost per task, health, verdict), and under Analytics → Governance the guardrail events (e.g. PII redacted from the HubSpot response before the LLM ever saw it). Every LLM and MCP call is a row in Operations → Logs & Audit → Proxy Logs with tokens, cost, latency, the arguments preview and result size — and the approval decision, who made it, and the retried call are linked there in one call chain.

The agent code contains none of this. That’s the point.

Related

• Your AI System — the same agent’s journey from here through deploy, govern, observe and assure
• Use MCP tools through the gateway — connecting other MCP clients
• Agent Skills — package agent behaviors as governed, versioned skills
• A2A agents — agent-to-agent messaging through the gateway
• Approvals — configuring which tools require a human