Xquik MCP Server with a 10xGraph Agent

"""Discover Xquik tools through the 10xGraph MCP adapter.

Run: XQUIK_API_KEY=... python examples/mcp/xquik/client.py
"""

import asyncio
import os

from fastmcp import Client

from tenxgraph.core.graph import ToolNode


MCP_URL = "https://xquik.com/mcp"


def require_api_key() -> str:
    """Read the Xquik API key without embedding it in source."""
    api_key = os.getenv("XQUIK_API_KEY", "").strip()
    if not api_key:
        raise RuntimeError("Set XQUIK_API_KEY before running this example")
    return api_key


def build_client(api_key: str) -> Client:
    """Build a Streamable HTTP client for the remote Xquik MCP server."""
    return Client(
        {
            "mcpServers": {
                "xquik": {
                    "url": MCP_URL,
                    "transport": "streamable-http",
                    "headers": {"x-api-key": api_key},
                }
            }
        }
    )


async def main() -> None:
    """Load the remote tool schemas through 10xGraph."""
    tool_node = ToolNode([], client=build_client(require_api_key()))
    tools = await tool_node.all_tools()
    names = sorted(tool["function"]["name"] for tool in tools)
    print(f"Available Xquik tools: {', '.join(names)}")


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

"""Give a ReactAgent the Xquik MCP tools.

Reuses build_client() from client.py. Run from the repo root:
XQUIK_API_KEY=... ANTHROPIC_API_KEY=... python examples/mcp/xquik/agent.py
"""

import asyncio

from client import build_client, require_api_key
from dotenv import load_dotenv

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


load_dotenv()


async def main() -> None:
    agent = ReactAgent(
        model="claude-sonnet-5-5",
        system_prompt=[{"role": "system", "content": "Use the Xquik tools to answer."}],
        client=build_client(require_api_key()),
    )
    app = agent.compile()
    result = await app.ainvoke(
        {"messages": [Message.text_message("Which Xquik tools are available to you?")]},
        config={"thread_id": "xquik-demo", "recursion_limit": 10},
    )
    print("assistant:", result["messages"][-1].text())
    await app.aclose()


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