​

MCP Tools

​

Your first MCP server and agent

# notes_server.py
from mcp.server.mcpserver import MCPServer

server = MCPServer("notes")

NOTES = {
    "wind": "Offshore wind capacity grew 20% in 2025.",
    "solar": "Rooftop solar installs doubled in Lagos.",
}


@server.tool()
def search_notes(query: str) -> str:
    """Return the notes whose topic appears in the query."""
    found = [text for topic, text in NOTES.items() if topic in query.lower()]
    return "\n".join(found) or "No notes."


@server.tool()
def note_stats() -> dict:
    """How many notes there are, and their topics."""
    return {"count": len(NOTES), "topics": sorted(NOTES)}


if __name__ == "__main__":
    server.run()

import asyncio
import sys
from omnicoreagent import OmniCoreAgent

async def main():
    agent = OmniCoreAgent(
        name="researcher",
        system_instruction="Use your tools to answer. Answer in plain text.",
        model_config={"provider": "openai", "model": "gpt-5.4-mini"},
        mcp_tools=[
            {
                "name": "notes",
                "transport_type": "stdio",
                "command": sys.executable,
                "args": ["notes_server.py"],
            }
        ],
    )
    await agent.connect_mcp_servers()

    result = await agent.run("How many notes are there, and what do they say about wind?")
    print(result["response"])

    trajectory = await agent.get_trajectory(result["trace_id"])
    for step in trajectory["steps"]:
        for call in step["tool_calls"]:
            print("step", step["step"], "->", call["provider"], call["server"], call["tool_name"], call["arguments"], call["outcome"])
    for server in trajectory["harness"]["mcp_servers"]:
        print(server["name"], server["status"], server["server_info"], server["tool_count"], "tools")

    await agent.cleanup()

asyncio.run(main())

There are 2 notes total.

About wind, the note says: “Offshore wind capacity grew 20% in 2025.”
step 1 -> mcp notes note_stats {} success
step 1 -> mcp notes search_notes {'query': 'wind'} success
notes connected {'name': 'notes', 'version': ''} 2 tools

​

How it works

• Connecting. connect_mcp_servers() starts every stdio server and connects every HTTP server at once. Each does the MCP handshake and lists all its tools (every page) within connect_timeout. A server that fails does not stop the others: its reason is recorded, and the agent runs without its tools. OmniServe connects on startup.
• The name is the identity. The name you configure is what routing, governance, telemetry and readiness use. What a server reports about itself (its name and version) is kept as metadata only, since the server controls it. A server without a name gets a stable one derived from its command or host, such as python_c51c4f; naming each server is clearer.
• Calling. A call goes to the server that offered the tool, under call_timeout if you set one and always under the agent’s tool_call_timeout. If two servers offer a tool with the same name, the model is shown each under its own unambiguous alias.
• Closing. agent.cleanup() (or cleanup_mcp_servers()) closes every connection and stops stdio server processes.

​

What the model receives

• Structured content, when the server returns it, is the result. A text block that only repeats it is dropped; other blocks, such as an image, are kept beside it. A scalar the server wrapped as {"result": value} is returned as the value. A Python server sends structured content when the tool’s return type is typed (a TypedDict, a Pydantic model, dict[str, int], int); note_stats above is annotated as a bare dict, so the SDK sent it as JSON text instead.
• A single text block is returned as text; several blocks as a list, with their MCP field names (such as mimeType).
• A tool that reports failure (isError) is an error result.
• A protocol error is an error result that keeps its JSON-RPC code, such as MCP error -32602: ... for invalid arguments, or MCP error -32001 when call_timeout expires. The code is also in the result’s data as mcp_error.

​

When a connection drops

​

Remote servers: Streamable HTTP and SSE

from omnicoreagent import OmniCoreAgent

agent = OmniCoreAgent(
    name="researcher",
    system_instruction="Use your tools to answer. Answer in plain text.",
    model_config={"provider": "openai", "model": "gpt-5.4-mini"},
    mcp_tools=[
        {
            "name": "notes",
            "transport_type": "streamable_http",
            "url": "http://127.0.0.1:8765/mcp",
            "headers": {"Authorization": "Bearer dev-token"},
            "call_timeout": 20,
        }
    ],
)

{'name': 'notes', 'transport_type': 'streamable_http', 'status': 'connected', 'server_info': {'name': 'notes', 'version': ''}, 'protocol_version': '2025-11-25', 'tool_count': 2, 'reconnects': 0, 'error': None}

​

OAuth

server = {
    "name": "reports",
    "transport_type": "streamable_http",
    "url": "http://localhost:8000/mcp",
    "auth": {"method": "oauth", "callback_port": 3030},
}

​

Options

​

Governance

import asyncio
import sys
from omnicoreagent import OmniCoreAgent
from omnicoreagent.governance import PolicyEffect, PolicyRule, build_default_policy

policy = build_default_policy("interactive-dev")
policy.rules.ask = [
    rule for rule in policy.rules.ask
    if rule.rule_id not in {"ask_mcp_server_start", "ask_mcp_tool_call"}
]
policy.rules.allow += [
    PolicyRule(rule_id="start_notes", effect=PolicyEffect.ALLOW,
               capability="mcp.server.*", target={"mcp_server": "notes"}),
    PolicyRule(rule_id="call_notes", effect=PolicyEffect.ALLOW,
               capability="tool.mcp.call", target={"mcp_server": "notes"}),
]

async def main():
    agent = OmniCoreAgent(
        name="researcher",
        system_instruction="Use your tools to answer. Answer in plain text, in one sentence.",
        model_config={"provider": "openai", "model": "gpt-5.4-mini"},
        mcp_tools=[{"name": "notes", "command": sys.executable, "args": ["notes_server.py"]}],
        agent_config={"governance_config": {"enabled": True, "policy": policy}},
    )
    await agent.connect_mcp_servers()
    result = await agent.run("What do the notes say about solar?")
    print(result["status"], "-", result["response"])

    trajectory = await agent.get_trajectory(result["trace_id"])
    for step in trajectory["steps"]:
        for call in step["tool_calls"]:
            decision = call["governance"][0]
            print(call["tool_name"], decision["capability"], decision["effect"], decision["matched_rule_ids"])
    await agent.cleanup()

asyncio.run(main())

success - The notes say: “Rooftop solar installs doubled in Lagos.”
search_notes tool.mcp.call allow ['call_notes']

​

Readiness and telemetry

• The run’s header (trajectory["harness"]["mcp_servers"]) lists every configured server with its status (connected, disconnected, failed or not_connected), reported name and version, protocol version, tool count, reconnects and last error. It never includes commands, arguments, environment, URLs or headers, which can carry credentials.
• Each MCP call in the trajectory names its provider (mcp) and server, and lists any reconnect. See Observability.
• OmniServe’s GET /ready reports mcp_connected (every configured server connected) and mcp_servers, each server’s status and last error, so one failed server among several is visible (OmniServe).

​

When things go wrong

success - I don’t see any notes in the workspace yet, so I can’t tell what they say about wind.
[{'name': 'notes', 'transport_type': 'stdio', 'status': 'failed', 'server_info': None, 'protocol_version': None, 'tool_count': 0, 'reconnects': 0, 'error': 'Connection closed'}]

ValueError: MCP server 'notes': command is required for stdio transport
ValueError: MCP server 'notes': 'headers' does not apply to stdio (it applies to sse, streamable_http)
ValueError: MCP server 'notes': Unknown MCP server setting 'arg'. Allowed: args, auth, call_timeout, command, connect_timeout, cwd, env, headers, name, sse_read_timeout, timeout, transport_type, url
ValueError: MCP server 'notes': url must start with http:// or https://
ValueError: MCP server 'notes': Unsupported MCP transport_type 'websocket'. Supported: stdio, sse, streamable_http

​

Next