Integrations

Adapters that bridge Locus to external frameworks.

FastMCP

Expose a Locus agent (or any of its tools) as a Model Context Protocol server, and consume MCP tools from any compliant client as native Locus Tools.

LocusMCPServer

Bases: BaseModel

Exposes a Locus Agent as an MCP server.

This allows Locus agents to be used by any MCP-compatible client.

Example

from locus import Agent from locus.integrations import LocusMCPServer

agent = Agent(model=model, tools=[...]) server = LocusMCPServer(agent=agent, name="my-agent") server.run() # Starts MCP server

run

run(transport: Literal['stdio', 'http', 'sse', 'streamable-http'] = 'stdio') -> None

Run the MCP server.

Parameters:

Name	Type	Description	Default
transport	Literal['stdio', 'http', 'sse', 'streamable-http']	Transport type ("stdio", "http", "sse", or "streamable-http").	'stdio'

Source code in src/locus/integrations/fastmcp.py

def run(self, transport: Literal["stdio", "http", "sse", "streamable-http"] = "stdio") -> None:
    """
    Run the MCP server.

    Args:
        transport: Transport type ("stdio", "http", "sse", or "streamable-http").
    """
    if self._mcp is None:
        self._mcp = self._create_mcp()

    self._mcp.run(transport=transport)

handle_request async

handle_request(request: dict[str, Any]) -> dict[str, Any]

Handle a single MCP request (for testing).

Source code in src/locus/integrations/fastmcp.py

async def handle_request(self, request: dict[str, Any]) -> dict[str, Any]:
    """Handle a single MCP request (for testing)."""
    if self._mcp is None:
        self._mcp = self._create_mcp()

    # Process based on method
    method = request.get("method", "")

    if method == "tools/list":
        tools = []
        if hasattr(self.agent, "_tool_registry"):
            self.agent._initialize()
            for tool_obj in self.agent._tool_registry.tools.values():
                tools.append(locus_tool_to_mcp(tool_obj))
        return {"tools": tools}

    if method == "tools/call":
        params = request.get("params", {})
        tool_name = params.get("name", "")
        arguments = params.get("arguments", {})

        if tool_name == "run_agent":
            result = self.agent.run_sync(arguments.get("prompt", ""))
            return {"content": [{"type": "text", "text": result.message}]}

        # Find and execute the tool
        if hasattr(self.agent, "_tool_registry"):
            self.agent._initialize()
            tool_obj = self.agent._tool_registry.get(tool_name)
            if tool_obj:
                result = await tool_obj.execute(**arguments)
                text = result if isinstance(result, str) else json.dumps(result)
                return {"content": [{"type": "text", "text": text}]}

        return {"error": {"code": -32602, "message": f"Unknown tool: {tool_name}"}}

    return {"error": {"code": -32601, "message": f"Unknown method: {method}"}}

create_mcp_server

create_mcp_server(agent: Agent, name: str = 'locus-agent', version: str = '1.0.0') -> LocusMCPServer

Create an MCP server from a Locus Agent.

Parameters:

Name	Type	Description	Default
agent	Agent	Locus Agent instance	required
name	str	Server name	'locus-agent'
version	str	Server version	'1.0.0'

Returns:

Type	Description
LocusMCPServer	LocusMCPServer instance

Example

server = create_mcp_server(agent, name="my-assistant") server.run()

Source code in src/locus/integrations/fastmcp.py

def create_mcp_server(
    agent: Agent,
    name: str = "locus-agent",
    version: str = "1.0.0",
) -> LocusMCPServer:
    """
    Create an MCP server from a Locus Agent.

    Args:
        agent: Locus Agent instance
        name: Server name
        version: Server version

    Returns:
        LocusMCPServer instance

    Example:
        >>> server = create_mcp_server(agent, name="my-assistant")
        >>> server.run()
    """
    return LocusMCPServer(agent=agent, name=name, version=version)

mcp_tool_to_locus

mcp_tool_to_locus(name: str, description: str, func: Callable[..., Any], parameters: dict[str, Any] | None = None) -> Tool

Convert an MCP-style tool to a Locus Tool.

When parameters is provided, the JSON Schema is used as-is to construct the Tool. This preserves the source tool's flat-field schema end-to-end so the LLM sees the original argument shape (e.g. {tenant_id, regex, limit}) instead of the generic {kwargs: …} shape that the @tool decorator would otherwise derive from the wrapper's **kwargs signature.

When parameters is omitted, falls back to the decorator-derived schema (parameter-less tool) for backward compatibility.

Parameters:

Name	Type	Description	Default
name	str	Tool name	required
description	str	Tool description	required
func	Callable[..., Any]	The async function to call	required
parameters	dict[str, Any] | None	JSON Schema for parameters	None

Returns:

Type	Description
Tool	Locus Tool instance

Source code in src/locus/integrations/fastmcp.py

def mcp_tool_to_locus(
    name: str,
    description: str,
    func: Callable[..., Any],
    parameters: dict[str, Any] | None = None,
) -> Tool:
    """
    Convert an MCP-style tool to a Locus Tool.

    When ``parameters`` is provided, the JSON Schema is used **as-is** to
    construct the Tool. This preserves the source tool's flat-field
    schema end-to-end so the LLM sees the original argument shape
    (e.g. ``{tenant_id, regex, limit}``) instead of the generic
    ``{kwargs: …}`` shape that the ``@tool`` decorator would otherwise
    derive from the wrapper's ``**kwargs`` signature.

    When ``parameters`` is omitted, falls back to the decorator-derived
    schema (parameter-less tool) for backward compatibility.

    Args:
        name: Tool name
        description: Tool description
        func: The async function to call
        parameters: JSON Schema for parameters

    Returns:
        Locus Tool instance
    """

    async def _invoke(**kwargs: Any) -> str:
        result = await func(**kwargs)
        if isinstance(result, str):
            return result
        return json.dumps(result)

    if parameters is not None:
        # Direct construction: keep the source MCP server's
        # inputSchema as the Tool's parameters dict. The Tool's
        # execute path forwards ``**kwargs`` to ``_invoke`` which
        # forwards them to the original ``func``, so the LLM's tool
        # call args land flat at the server.
        return Tool(
            name=name,
            description=description,
            parameters=parameters,
            fn=_invoke,
            idempotent=False,
        )

    # Fallback: derive the schema from the wrapper signature.
    # No-args tools work; tools that need typed args should pass
    # ``parameters=`` explicitly.
    @tool(name=name, description=description)
    async def wrapper(**kwargs: Any) -> str:
        return await _invoke(**kwargs)

    return wrapper