Skip to content

MCP Guide

This guide explains how kneo_agent integrates with MCP tool servers.

What MCP Means In Kneo Agent

Kneo Agent treats an MCP server as an external tool source.

The flow is:

  1. Connect to an MCP server with MCPServerConfig
  2. Discover remote tools with ToolRegistry.register_mcp_server(...)
  3. Convert those tools into normal ToolDefinition objects plus proxy handlers
  4. Package the registry into an agent with AgentBuilder.with_tool_registry(...)

That means MCP support does not require a new runtime type. Bridge, Native, and Adapter runtimes can all use MCP tools through the existing RunConfig and tool_handlers path.

Middleware composes with MCP naturally:

  • use run middleware for tracing, guardrails, or request shaping
  • use Bridge tool-call middleware when you want to inspect or rewrite MCP tool results before they are fed back into the loop

Supported Transports

Kneo Agent supports three MCP connection styles:

  • stdio: launch a subprocess and speak JSON-RPC over standard I/O
  • http: send JSON-RPC requests to an HTTP endpoint
  • sse: listen on an SSE stream and send JSON-RPC requests to a message endpoint

Stdio Example

import asyncio
from kneo_agent import AgentBuilder, MCPServerConfig
from kneo_agent.patterns import NativeRuntimeFactory
from kneo_agent.utils import ToolRegistry


async def main() -> None:
    registry = ToolRegistry()
    await registry.register_mcp_server(
        MCPServerConfig.stdio(
            name="filesystem",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
        ),
        prefix="fs_",
    )

    runtime = NativeRuntimeFactory.for_openai(model="gpt-4o", strategy="react")
    agent = (
        AgentBuilder()
        .with_name("Filesystem Assistant")
        .with_tool_registry(registry, skill_name="filesystem-mcp")
        .use_runtime(runtime)
        .build()
    )

    print(await agent.chat("List the files in /tmp."))
    await registry.aclose()


asyncio.run(main())

HTTP Example

config = MCPServerConfig.http(
    name="remote-http",
    url="https://example.com/mcp",
    headers={"Authorization": "Bearer <token>"},
)

SSE Example

config = MCPServerConfig.sse(
    name="remote-sse",
    sse_url="https://example.com/mcp/events",
    message_url="https://example.com/mcp/messages",
)

For SSE, some servers publish the message endpoint through the event stream. If yours does, message_url can be omitted.

TLS / mTLS / custom CA (v1.2.0)

Both MCPServerConfig.http() and .sse() accept TLS knobs for on-prem deployments behind corporate certificate authorities or with mutual-TLS requirements:

config = MCPServerConfig.http(
    name="internal",
    url="https://mcp.internal.corp/v1",
    headers={"X-Internal-Auth": "..."},
    ca_bundle="/etc/ssl/corp/ca.pem",       # custom root CA
    client_cert="/etc/ssl/corp/client.crt", # client cert for mTLS
    client_key="/etc/ssl/corp/client.key",  # matching key
)
Parameter Purpose
verify Whether to verify the server's TLS certificate. Default True; set False only for development against a known-bad self-signed cert.
ca_bundle Path to a PEM-encoded CA bundle. Use this for corporate or internal CAs that aren't in the system trust store.
client_cert Path to a PEM-encoded client certificate for mTLS.
client_key Path to the client certificate's private key. Required when client_cert is set unless the cert file already contains the key.

The transports build an ssl.SSLContext from these knobs and thread it into urlopen(..., context=...). Apps that need to inspect or share the context can call MCPServerConfig.build_ssl_context() directly — it returns None when no TLS customization is requested (the default-trust-store case).

For loading a list of internal MCP servers from a YAML/JSON config at app startup, see examples/cookbook/mcp_server_catalog.py.

Working With ToolRegistry

ToolRegistry.register_mcp_server(...) returns a list of discovered ToolDefinition objects and registers callable proxies for each one.

tools = await registry.register_mcp_server(config, prefix="remote_")
print([tool.name for tool in tools])

Use prefix when you want to namespace a server's tools locally and avoid name collisions with other MCP servers or local Python tools.

When you are done, close the registry to clean up subprocesses and sessions:

await registry.aclose()

Working With AgentBuilder

The easiest way to expose MCP tools to an agent is:

agent = (
    AgentBuilder()
    .with_name("MCP Assistant")
    .with_tool_registry(registry, skill_name="mcp-tools")
    .use_runtime(runtime)
    .build()
)

with_tool_registry(...) does two things:

  • copies registry.definitions into the agent's tool list
  • packages the registry handlers into an implicit Skill

That keeps MCP integration aligned with the existing skill/tool resolution pipeline.