MCP Tools in a Python AI Agent with 10xGraph

"""A small FastMCP weather server over streamable HTTP.

Run it in its own terminal first:
    python examples/mcp/local/server.py                      # port 8000
    python examples/mcp/local/server.py --port 8001 --name eu  # a second server
"""

import argparse

from fastmcp import Context, FastMCP


def build_server(name: str) -> FastMCP:
    mcp = FastMCP(f"{name} weather server")

    # With pass_user_info_to_mcp=True (see agent.py) the ToolNode sends the caller as request
    # metadata. A `Context` parameter is injected by FastMCP and never shows in the tool
    # schema the LLM sees, so the model cannot invent or change the caller.
    @mcp.tool(
        description="Get the current weather for a location",
        tags={"weather", "information"},
    )
    def get_weather(location: str, ctx: Context) -> dict:
        # ToolNode sends the caller as request metadata under the key "user".
        meta = ctx.request_context.meta
        user = (meta.model_extra or {}).get("user") if meta else None
        print(f"[{name}] weather for {location}, caller: {user}")
        return {"location": location, "temperature": "22C", "description": f"Sunny ({name})"}

    @mcp.tool(description="Get a 3-day forecast for a location")
    def get_forecast(location: str) -> list[dict]:
        return [{"day": d, "temperature": f"{20 + d}C", "description": "Clear"} for d in range(3)]

    return mcp


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--name", default="us")
    parser.add_argument("--port", type=int, default=8000)
    args = parser.parse_args()
    build_server(args.name).run(transport="streamable-http", host="127.0.0.1", port=args.port)

"""Talk to an MCP server directly: list its tools and call one, with no LLM involved.

Useful to check a server before giving it to an agent.
Start `python examples/mcp/local/server.py` first, then run this file.
"""

import asyncio

from dotenv import load_dotenv
from fastmcp import Client


load_dotenv()

config = {
    "mcpServers": {
        "weather": {
            "url": "http://127.0.0.1:8000/mcp",
            "transport": "streamable-http",
            # Remote servers usually want a header like this one.
            "headers": {"Authorization": "Bearer TEST_WEATHER_API_KEY"},
        },
    },
}


async def main() -> None:
    async with Client(config) as client:
        for tool in await client.list_tools():
            tags = (tool.meta or {}).get("_fastmcp", {}).get("tags", [])
            print(f"tool: {tool.name} tags={tags}")
            print("  schema:", tool.inputSchema)

        result = await client.call_tool("get_weather", {"location": "New York"})
        print("data:", result.data)


if __name__ == "__main__":
    asyncio.run(main())

"""A ReactAgent that uses tools from two local MCP servers.

Start two servers first (separate terminals):
    python examples/mcp/local/server.py --name us --port 8000
    python examples/mcp/local/server.py --name eu --port 8001
Then: GEMINI_API_KEY=... python examples/mcp/local/agent.py
"""

import asyncio

from dotenv import load_dotenv
from fastmcp import Client

from tenxgraph.core.state import Message
from tenxgraph.prebuilt.agent import ReactAgent


load_dotenv()

# Each server needs its own URL. With more than one server, fastmcp prefixes every tool
# with the server key, so the agent sees `us_get_weather` and `eu_get_weather`
# (a single-server config keeps the plain name).
mcp_config = {
    "mcpServers": {
        "us": {"url": "http://127.0.0.1:8000/mcp", "transport": "streamable-http"},
        "eu": {"url": "http://127.0.0.1:8001/mcp", "transport": "streamable-http"},
    }
}

agent = ReactAgent(
    model="gemini-2.5-flash",
    system_prompt=[
        {"role": "system", "content": "Answer weather questions with the available tools."}
    ],
    client=Client(mcp_config),
    # Sends config["user"] (or {"user_id": ...}) with every MCP call as request metadata (meta["user"]).
    pass_user_info_to_mcp=True,
)


async def main() -> None:
    app = agent.compile()
    result = await app.ainvoke(
        {"messages": [Message.text_message("Compare the weather in New York and Berlin.")]},
        config={"thread_id": "mcp-local", "user_id": "alice", "recursion_limit": 10},
    )
    print("assistant:", result["messages"][-1].text())
    await app.aclose()


if __name__ == "__main__":
    asyncio.run(main())