Skip to content

MCP servers

With ParaLLeM, you can use local MCP servers with the Batch API. Use the pllm.MCPOutput object.

examples/mcp/example_local_mcp.py
import parallem as pllm
from fastmcp import Client
from parallem.core.cast.convert_mcp import mcp_tool_to_tool_schema

mcp_client = Client("examples/mcp/math_server.py")

async def my_agent(agt: pllm.AgentContext, mcp_client: Client, question: str):
    conv = agt.get_msg_state()

    available_tools = await mcp_client.list_tools()
    schema = [mcp_tool_to_tool_schema(tool) for tool in available_tools]
    last_msg = conv.ask_llm(
        question,
        tools=schema,
    )

    while last_msg.function_calls:
        print(last_msg.function_calls)

        for fc in last_msg.function_calls:
            out = await mcp_client.call_tool(fc.name, fc.args)
            conv.append(
                pllm.MCPOutput(
                    name=fc.name,
                    call_id=fc.call_id,
                    content=out.content,
                )
            )

        last_msg = conv.ask_llm()
        print(conv[-1].final_answer)

async def main():
    with pllm.resume_directory(
        ".pllm/simplest",
        provider="anthropic",
        strategy="sync",
        dashboard=True,
        load_dotenv=True,
    ) as orch:
        with orch.agent() as agt:
            async with mcp_client:
                await my_agent(agt, mcp_client, "Add 3 and 6.")

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())
examples/mcp/math_server.py
from fastmcp import FastMCP

mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """Multiply two numbers"""
    return a * b

if __name__ == "__main__":
    mcp.run(transport="stdio")

Just run examples/mcp/example_local_mcp.py.

Warning

If MCP outputs are non-deterministic (ie. file system change), then there could be a cache miss, leading to disruption to batched multi-step workflows. Use the MessageState.save, MessageState.load, and memoize for non-deterministic blocks.

Server-side MCP

Some providers offer MCP servers as a server-side tool. In that case, use pllm.tools.MCPServerTool.

examples/mcp/example_server_mcp.py
import parallem as pllm

with pllm.resume_directory(
    ".pllm/simple/mcp",
    strategy="sync",
    provider="openai",
    dashboard=True,
    load_dotenv=True,
) as orch:
    with orch.agent() as agt:
        resp = agt.ask_llm(
            "Roll 2d4+1.",
            tools=[
                pllm.tools.MCPServerTool(
                    server_label="dmcp",
                    server_description="A Dungeons and Dragons MCP server to assist with dice rolling.",
                    server_url="https://dmcp-server.deno.dev/sse",
                    require_approval="never",
                )
            ],
        )

        print(resp.final_answer)

Examples