server

server

MCP Server — wraps OpenJarvis tools as MCP-discoverable tools.

Classes

MCPServer

MCPServer(tools: Optional[List[BaseTool]] = None, *, bus: Optional[EventBus] = None, capability_policy: Optional[Any] = None, rate_limiter: Optional[Any] = None, agent_id: str = _MCP_AGENT_ID)

MCP server that exposes OpenJarvis tools via JSON-RPC.

PARAMETER	DESCRIPTION
tools	List of BaseTool instances to expose. If None, auto-discovers all registered tools from ToolRegistry. TYPE: Optional[List[BaseTool]] DEFAULT: None

Create an MCP server with the same dispatch gates as other agents.

agent_id is deliberately non-empty by default: capability policy, rate-limit and audit records must never collapse MCP traffic into the anonymous "" identity. Callers that enable a security profile pass its policy and limiter here; ToolExecutor then applies both declared and canonical built-in capability requirements before a tool can run.

Source code in src/openjarvis/mcp/server.py

def __init__(
    self,
    tools: Optional[List[BaseTool]] = None,
    *,
    bus: Optional[EventBus] = None,
    capability_policy: Optional[Any] = None,
    rate_limiter: Optional[Any] = None,
    agent_id: str = _MCP_AGENT_ID,
) -> None:
    """Create an MCP server with the same dispatch gates as other agents.

    ``agent_id`` is deliberately non-empty by default: capability policy,
    rate-limit and audit records must never collapse MCP traffic into the
    anonymous ``""`` identity.  Callers that enable a security profile
    pass its policy and limiter here; ``ToolExecutor`` then applies both
    declared and canonical built-in capability requirements before a tool
    can run.
    """
    if tools is None:
        tools = self._auto_discover_tools()
    self._tools: Dict[str, BaseTool] = {t.spec.name: t for t in tools}
    self._executor = ToolExecutor(
        tools,
        bus,
        capability_policy=capability_policy,
        rate_limiter=rate_limiter,
        agent_id=agent_id or _MCP_AGENT_ID,
    )

Functions

get_tools

get_tools() -> List[BaseTool]

Return all tool instances (for use by SystemBuilder).

Source code in src/openjarvis/mcp/server.py

def get_tools(self) -> List[BaseTool]:
    """Return all tool instances (for use by SystemBuilder)."""
    return list(self._tools.values())

handle

handle(request: MCPRequest) -> MCPResponse

Dispatch an MCP request and return a response.

Source code in src/openjarvis/mcp/server.py

def handle(self, request: MCPRequest) -> MCPResponse:
    """Dispatch an MCP request and return a response."""
    if request.method == "initialize":
        return self._handle_initialize(request)
    elif request.method == "tools/list":
        return self._handle_tools_list(request)
    elif request.method == "tools/call":
        return self._handle_tools_call(request)
    else:
        return MCPResponse.error_response(
            request.id,
            METHOD_NOT_FOUND,
            f"Unknown method: {request.method}",
        )