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 hop 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 hop in Mission Control

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

Prerequisites

• The gateway running (see Quickstart)
• One or more MCP servers registered and assigned to a resource group (see Register your own MCP server)
• An API key for that resource group (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")

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.

What you get for free

Run the agent, then open Mission Control:

• Every LLM call with tokens, cost, latency — attributed to the API key’s resource group
• Every MCP tool call with its arguments preview and result size
• Guardrail events (e.g. PII redacted from the HubSpot response before the LLM ever saw it)
• The approval decision, who made it, and the retried call — linked in one call chain

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

Related

• 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