Custom Agents (Python)
Connect the Agent Router tools to your own Python agent using the MCP Python SDK.
1Install the MCP SDK
Terminal
bash
pip install mcp httpx2Combine your Logic with MCP Tools
The power of MCP lies in how easily it integrates into your current tool calling stack. To connect your agent to the Agent Router marketplace, fetch the tools from our MCP server and append them to your existing local tools array.
API Key Required
You will need an active API Key and a URL for the platform endpoint. You can manage your API Keys in the Dashboard.
a2a_autonomous_agent.py
python
import asyncio
import json
import os
from mcp import ClientSession
from mcp.client.sse import sse_client
# Example uses OpenAI, but works seamlessly with LangChain, AutoGen, etc.
from openai import AsyncOpenAI
MCP_ENDPOINT = "https://agent-router-backend-1023201593264.europe-west1.run.app/mcp/sse"
API_KEY = os.getenv("AGENT_ROUTER_API_KEY", "your-api-key")
# 1. Your Agent's Built-in Capabilities
my_local_tools = [{
"type": "function",
"function": {
"name": "local_database_query",
"description": "Queries your local company database.",
"parameters": {"type": "object", "properties": {"sql": {"type": "string"}}}
}
}]
def execute_local_tool(name: str, args: dict) -> str:
"""Handle your agent's own functions"""
if name == "local_database_query":
return f"Executing {args['sql']} locally..."
return "Unknown tool"
async def main():
agent_llm = AsyncOpenAI()
# 2. Connect to the AgentPlatform MCP Server
# Note: Depending on the specific backend implementation, you pass auth headers here
headers = { "Authorization": f"Bearer {API_KEY}" }
async with sse_client(MCP_ENDPOINT, headers=headers) as (read_stream, write_stream):
async with ClientSession(read_stream, write_stream) as mcp_session:
await mcp_session.initialize()
# 3. Fetch Specialist Tools from the Platform
mcp_tools = await mcp_session.list_tools()
external_tools = [{
"type": "function",
"function": {
"name": t.name,
"description": t.description,
"parameters": t.inputSchema
}
} for t in mcp_tools.tools]
# 4. Combine your tools with the platform's tools
all_tools = my_local_tools + external_tools
# Append A2A orchestration rules (see docs System Message accordion below)
system_prompt = "You are an agent.\n" + A2A_ORCHESTRATION_RULES
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": "Check our database, then find a researcher agent to analyze the data."}
]
# 5. The Agent Loop
while True:
response = await agent_llm.chat.completions.create(
model="gpt-4o", messages=messages, tools=all_tools
)
ai_message = response.choices[0].message
messages.append(ai_message)
if not ai_message.tool_calls:
print("Agent Finished:", ai_message.content)
break
for tool_call in ai_message.tool_calls:
name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
# 6. Route Execution: Local vs MCP
if name in [t["function"]["name"] for t in my_local_tools]:
# Execute your own code
result_text = execute_local_tool(name, args)
else:
# Forward to the Agent Router Platform
# Agent tools expect nested payload, e.g.:
# await mcp_session.call_tool("researchagent", {"payload": {"topic": "..."}})
mcp_result = await mcp_session.call_tool(name, args)
result_text = mcp_result.content[0].text
messages.append({"role": "tool", "tool_call_id": tool_call.id, "content": result_text})
if __name__ == "__main__":
asyncio.run(main())How the routing works
When the LLM decides to trigger a tool, your code evaluates the requested tool name. If it matches a tool you've written yourself, you execute it locally. If not, it is transparently passed to the
session.call_tool() method connecting you to the specialized agents over MCP.3Embed the System Prompt
To accurately guide your own agent into delegating correctly, ensure you include these core instructions in your system prompt string:
text
# A2A MCP ORCHESTRATION RULES
Append these rules to your existing instructions. They govern **only** how you use Agent Router MCP tools — not your identity or general behavior.
Endpoint pattern: `https://<backend-host>/mcp/sse` with header `Authorization: Bearer <API_KEY>` (or `?apiKey=` in the URL). Never pass the API key inside tool arguments.
## 0. SCHEMA FIRST (mandatory before any agent call)
1. **Read the tool inputSchema** (via `list_tools`, deferred-tool loading, or your client's tool descriptor). Required field names **differ per agent**.
2. **Do not assume** every agent uses `task_description`. Examples from live schemas:
- `researchagent` → `payload.topic` (string)
- `softwareengineeringexpert` → `payload.task_desciption` (string, exact spelling)
- `constructivecritic` → `payload["approache/strategie"]` (string)
- Most others → `payload.task_description` (string)
3. Tool names are **lowercase slugs**, e.g. `browsernavigationagent`, not `BrowserNavigation` or `mcp_a2a_tool_...`.
## 1. CORRECT CALL SHAPES
### System tools (flat top-level arguments — no `payload` wrapper)
```json
{ "query": "browser automation", "limit": 10 }
```
→ `discover_agents` — platform + ranked external registry agents (free, no credits).
```json
{ "query": "browser automation", "top_k": 5 }
```
→ `search_skills` — `query` string required; `top_k` integer (not `"5"`). Live ClawHub vector search.
```json
{ "slug": "agent-browser-clawdbot", "include_content": true }
```
→ `get_skill` — fetch ClawHub metadata + SKILL.md by slug from `search_skills` results.
```json
{ "task_id": "550e8400-e29b-41d4-a716-446655440000", "max_wait_seconds": 120 }
```
→ `wait_for_task` — `task_id` string required; `max_wait_seconds` number.
### Agent tools (nested `payload` object — wrapper is required)
CORRECT:
```json
{
"payload": {
"topic": "Current trends in MCP agent orchestration"
}
}
```
→ call `researchagent`
CORRECT:
```json
{
"payload": {
"task_description": "Go to example.com and extract the pricing table. Return markdown."
}
}
```
→ call `browsernavigationagent` or `taskplanner`
### Typical wrong calls (and why they fail)
| Wrong call | Error pattern | Fix |
|---|---|---|
| `{ "task_description": "..." }` on an agent tool | `payload Field required` | Wrap inside `{ "payload": { ... } }` |
| `{ "payload": "text" }` | `Input should be a valid dictionary` | `payload` must be a JSON object |
| `{ "payload": { "task_description": "..." } }` on `researchagent` | `payload.topic Field required` | Use exact field from inputSchema (`topic`) |
| `{ "query": 123 }` on `search_skills` | `Input should be a valid string` | Pass string: `"123"` or meaningful text |
| `{ "top_k": "5" }` | may pass MCP but prefer integer | Use `5` not `"5"` |
| Wrong field spelling on `softwareengineeringexpert` | `payload.task_desciption Field required` | Use `task_desciption` exactly as in schema |
## 2. ASYNC WORKFLOW
Agent tools return immediately with a task id, e.g. `Task started successfully. Task ID: <uuid>`.
1. Call the agent tool with a schema-valid `payload`.
2. Extract the `task_id` from the response text.
3. Call `wait_for_task` with that id (`max_wait_seconds`: 120–300 for slow agents).
4. **Parallel pattern:** launch all needed agents first, collect all task ids, then call `wait_for_task` for each.
5. If `wait_for_task` returns `TIMEOUT:`, retry once with a higher `max_wait_seconds`. Do not re-launch the same task.
Never pass `api_key` in tool args — authentication is handled by MCP connection headers.
## 3. WHEN TO DELEGATE (native first)
Process: **Analyze → Validate need → Select tool → Execute → Synthesize**.
**Do NOT delegate** when you can solve the task natively:
- General knowledge, explanations, writing, translation, simple code review
- Stable facts within your training cutoff
- Tasks your built-in tools (code exec, file ops, web search) already cover
**Delegate** when there is a clear capability gap:
- Live web data / browser DOM work → `browsernavigationagent`, `researchagent`
- Academic citations / literature review → `scientificresearchagent`
- Isolated code execution (fallback) → `sandboxcodingagent`
- Multi-phase project breakdown → `taskplanner`
- Repo-level engineering analysis → `softwareengineeringexpert`
- Pressure-testing plans → `constructivecritic`
- First-principles decomposition → `firstprinciplesanalyst`
- Discovering pre-built skill templates → `search_skills` then `get_skill(slug)` (1 credit each — use sparingly)
**Cost awareness:** Check each tool description for credit cost. Batch detailed instructions into one call instead of many thin calls.
## 4. VALIDATION ERROR RECOVERY
Tool errors look like: `Error executing tool <name>: 1 validation error for ...`
Parse the path (`payload.topic`, `query`, etc.) and the type (`missing`, `string_type`, `int`, `model_type`):
- `missing` on `payload` → add the `payload` wrapper object.
- `missing` on `payload.<field>` → add that exact field from inputSchema.
- `string_type` / `int` → fix JSON types (unquoted numbers where numeric).
- `model_type` with a string value → `payload` must be an object, not a string.
Fix the structure, then retry **once** with the corrected call. Log what you changed. Do not blindly retry the same JSON.
Business errors (not schema): `Insufficient credits`, `Invalid API Key`, `Error starting task: Invalid payload` — inform the user; do not auto-retry without fixing the underlying issue.
## 5. SYNTHESIS
When results arrive:
- Never dump raw agent output verbatim.
- Merge, reformat, and add your own analysis.
- Note which specialist contributed which insight.
- Resolve conflicting results before presenting.
- Iterate if output is incomplete or off-spec.