Agent frameworks

An agent built with LangGraph, ADK or plain Python reaches its tools through MCP clients. To have each tool call checked, point those clients to the MCP URLs we give you and send our access token in the Authorization header. The framework stays as it is: only the address and the header change.

This page goes from the setup in the Admin to the first checked call.

What the administrator preparesCopy Anchor Link

The setup happens in the Admin, often by someone other than the developer, in this order:

1. An identity provider is connected, so that we can verify the person the agent acts for.
2. Each MCP server the agent uses is registered and its tools are reviewed (Customer servers, Tool review).
3. The agent exists, with the people or services that delegate to it and the tools it can call, and its access is published (Create an agent).

The developer then receives three things:

• The key file, <agent>.chokepoint.json. It comes from Generate key and download, in the Identity tab of the agent (Agent identity). The file holds the private key of the agent, so keep it in a secret store.
• The MCP URL of each server, listed in the same tab under MCP URL for developers, in the form <origin>/api/v1/mcp/<workspace>/<environment>/<server_id>.
• The token of the person the agent works for. Your identity provider issues it when that person signs in to your application. A job that runs with no person uses the token of a service account that delegates to the agent (Who can delegate).

Get an access tokenCopy Anchor Link

The agent does not send the identity provider token to the tools. It trades that token for ours and signs the request with the key from the file: an OAuth token exchange (RFC 8693) authenticated by a signed client assertion (RFC 7523). In Python, with requests and pyjwt[crypto]:

import json, time, uuid

import jwt
import requests

agent = json.load(open("issues-agent.chokepoint.json"))

def exchange_token(person_token: str) -> str:
    now = int(time.time())
    assertion = jwt.encode(
        {"iss": agent["client_id"], "sub": agent["client_id"], "aud": agent["token_endpoint"],
         "jti": str(uuid.uuid4()), "iat": now, "exp": now + 60},
        jwt.PyJWK(agent["private_key"]).key,
        algorithm="ES256",
        headers={"kid": agent["kid"]},
    )
    response = requests.post(agent["token_endpoint"], data={
        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
        "subject_token": person_token,
        "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
        "client_id": agent["client_id"],
        "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
        "client_assertion": assertion,
    })
    response.raise_for_status()
    return response.json()["access_token"]

Each assertion needs a new jti: a reused one is refused. The access token lasts at most 5 minutes.

A refused exchange answers invalid_client when the key does not match the agent, and invalid_grant when the person token is invalid or expired, or when the person does not delegate to the agent. Its error_description carries a Reference code. The response never says why; the administrator pastes the code in Quick incident to see the reason. Integrate the agent describes every field of the exchange.

Load the toolsCopy Anchor Link

The access token goes in the header of each MCP connection. An agent that uses several servers opens one connection per server, each with its own URL and the same token.

LangGraphCopy Anchor Link

LangGraph loads MCP tools with langchain-mcp-adapters:

from langchain_mcp_adapters.client import MultiServerMCPClient

token = exchange_token(person_token)

client = MultiServerMCPClient({
    "issues": {
        "transport": "streamable_http",
        "url": "<MCP URL of the issues server>",
        "headers": {"Authorization": f"Bearer {token}"},
    },
})
tools = await client.get_tools()

The tools list goes to the graph like any other LangChain tool.

ADKCopy Anchor Link

ADK loads MCP tools with McpToolset:

from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset

token = exchange_token(person_token)

issues = McpToolset(connection_params=StreamableHTTPConnectionParams(
    url="<MCP URL of the issues server>",
    headers={"Authorization": f"Bearer {token}"},
))
agent = LlmAgent(model="<model>", name="issues_agent", tools=[issues])

Tools that run inside the agentCopy Anchor Link

A tool written in the agent's own code, such as a @tool function in LangGraph or a function tool in ADK, runs in the agent process and never crosses an MCP connection, so we cannot check it. The same holds for an MCP server that the agent calls at its own address instead of the URL we give.

To bring such a tool under control, move it to an MCP server of your own:

1. Write the tool in a FastMCP server that validates our token (Validate the token).
2. Register the server in the Admin as a customer server and review its tools.
3. Load it in the agent like any other server, with its MCP URL.

The tool then runs only after the call is allowed, and its token tells it the person (sub) and the agent (act.sub).

Long runs and denied callsCopy Anchor Link

The examples set the token once, when they create the MCP client. A run that lasts more than 5 minutes needs a new exchange and a new client.

A denied tool call comes back as a JSON-RPC error with the message Access denied. Reference <code>. and the code in data.reference. Treat it as an answer, not as a network failure, and log the code (The tool calls).