Index

mcp

MCP (Model Context Protocol) layer for OpenJarvis.

Classes

MCPClient

MCPClient(transport: MCPTransport)

Client that communicates with an MCP server via a transport.

PARAMETER	DESCRIPTION
transport	The transport layer to use for communication. TYPE: MCPTransport

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

def __init__(self, transport: MCPTransport) -> None:
    self._transport = transport
    self._initialized = False
    self._capabilities: Dict[str, Any] = {}
    self._id_counter = itertools.count(1)

Functions

initialize

initialize() -> Dict[str, Any]

Perform the MCP initialize handshake.

Returns the server capabilities.

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

def initialize(self) -> Dict[str, Any]:
    """Perform the MCP initialize handshake.

    Returns the server capabilities.
    """
    response = self._send("initialize")
    self._initialized = True
    self._capabilities = response.result.get("capabilities", {})
    return response.result

list_tools

list_tools() -> List[ToolSpec]

Discover available tools from the server.

Returns a list of ToolSpec objects.

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

def list_tools(self) -> List[ToolSpec]:
    """Discover available tools from the server.

    Returns a list of ``ToolSpec`` objects.
    """
    response = self._send("tools/list")
    tools = response.result.get("tools", [])
    return [
        ToolSpec(
            name=t["name"],
            description=t.get("description", ""),
            parameters=t.get("inputSchema", {}),
        )
        for t in tools
    ]

call_tool

call_tool(name: str, arguments: Dict[str, Any] | None = None) -> Dict[str, Any]

Call a tool on the server.

Returns the result dictionary with content and isError fields.

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

def call_tool(
    self, name: str, arguments: Dict[str, Any] | None = None,
) -> Dict[str, Any]:
    """Call a tool on the server.

    Returns the result dictionary with ``content`` and ``isError`` fields.
    """
    response = self._send(
        "tools/call",
        {"name": name, "arguments": arguments or {}},
    )
    return response.result

close

close() -> None

Close the transport connection.

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

def close(self) -> None:
    """Close the transport connection."""
    self._transport.close()

MCPError dataclass

MCPError(code: int, message: str, data: Any = None)

Bases: Exception

MCP protocol error with JSON-RPC error code.

MCPNotification dataclass

MCPNotification(method: str, params: Dict[str, Any] = dict(), jsonrpc: str = '2.0')

JSON-RPC 2.0 notification (no id, no response expected).

Functions

to_json

to_json() -> str

Serialize to JSON string.

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

def to_json(self) -> str:
    """Serialize to JSON string."""
    return json.dumps(
        {
            "jsonrpc": self.jsonrpc,
            "method": self.method,
            "params": self.params,
        }
    )

MCPRequest dataclass

MCPRequest(method: str, params: Dict[str, Any] = dict(), id: int | str = 0, jsonrpc: str = '2.0')

JSON-RPC 2.0 request message.

Functions

to_json

to_json() -> str

Serialize to JSON string.

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

def to_json(self) -> str:
    """Serialize to JSON string."""
    return json.dumps(
        {
            "jsonrpc": self.jsonrpc,
            "id": self.id,
            "method": self.method,
            "params": self.params,
        }
    )

from_json classmethod

from_json(data: str) -> MCPRequest

Deserialize from JSON string.

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

@classmethod
def from_json(cls, data: str) -> MCPRequest:
    """Deserialize from JSON string."""
    parsed = json.loads(data)
    return cls(
        method=parsed["method"],
        params=parsed.get("params", {}),
        id=parsed.get("id", 0),
        jsonrpc=parsed.get("jsonrpc", "2.0"),
    )

MCPResponse dataclass

MCPResponse(result: Any = None, error: Optional[Dict[str, Any]] = None, id: int | str = 0, jsonrpc: str = '2.0')

JSON-RPC 2.0 response message.

Functions

to_json

to_json() -> str

Serialize to JSON string.

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

def to_json(self) -> str:
    """Serialize to JSON string."""
    obj: Dict[str, Any] = {"jsonrpc": self.jsonrpc, "id": self.id}
    if self.error is not None:
        obj["error"] = self.error
    else:
        obj["result"] = self.result
    return json.dumps(obj)

from_json classmethod

from_json(data: str) -> MCPResponse

Deserialize from JSON string.

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

@classmethod
def from_json(cls, data: str) -> MCPResponse:
    """Deserialize from JSON string."""
    parsed = json.loads(data)
    return cls(
        result=parsed.get("result"),
        error=parsed.get("error"),
        id=parsed.get("id", 0),
        jsonrpc=parsed.get("jsonrpc", "2.0"),
    )

error_response classmethod

error_response(id: int | str, code: int, message: str, data: Any = None) -> MCPResponse

Create an error response.

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

@classmethod
def error_response(
    cls,
    id: int | str,
    code: int,
    message: str,
    data: Any = None,
) -> MCPResponse:
    """Create an error response."""
    error: Dict[str, Any] = {"code": code, "message": message}
    if data is not None:
        error["data"] = data
    return cls(error=error, id=id)

MCPServer

MCPServer(tools: Optional[List[BaseTool]] = None)

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

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

def __init__(self, tools: Optional[List[BaseTool]] = None) -> None:
    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)

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}",
        )

InProcessTransport

InProcessTransport(server: MCPServer)

Bases: MCPTransport

Direct in-process transport for testing.

Routes requests directly to an MCPServer instance without serialization overhead.

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

def __init__(self, server: MCPServer) -> None:
    self._server = server

Functions

send

send(request: MCPRequest) -> MCPResponse

Dispatch request directly to the server.

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

def send(self, request: MCPRequest) -> MCPResponse:
    """Dispatch request directly to the server."""
    return self._server.handle(request)

close

close() -> None

No resources to release.

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

def close(self) -> None:
    """No resources to release."""

MCPTransport

Bases: ABC

Abstract transport layer for MCP communication.

Functions

send abstractmethod

send(request: MCPRequest) -> MCPResponse

Send a request and return the response.

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

@abstractmethod
def send(self, request: MCPRequest) -> MCPResponse:
    """Send a request and return the response."""

close abstractmethod

close() -> None

Release transport resources.

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

@abstractmethod
def close(self) -> None:
    """Release transport resources."""

SSETransport

SSETransport(url: str)

Bases: MCPTransport

JSON-RPC over HTTP with Server-Sent Events.

Sends requests via HTTP POST and reads SSE responses.

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

def __init__(self, url: str) -> None:
    self._url = url

Functions

send

send(request: MCPRequest) -> MCPResponse

Send request via HTTP POST.

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

def send(self, request: MCPRequest) -> MCPResponse:
    """Send request via HTTP POST."""
    import httpx

    response = httpx.post(
        self._url,
        json=json.loads(request.to_json()),
        headers={"Content-Type": "application/json"},
        timeout=30.0,
    )
    response.raise_for_status()
    return MCPResponse.from_json(response.text)

close

close() -> None

No persistent connection to close.

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

def close(self) -> None:
    """No persistent connection to close."""

StdioTransport

StdioTransport(command: List[str])

Bases: MCPTransport

JSON-RPC over stdin/stdout subprocess transport.

Launches a subprocess and communicates via JSON lines on stdin/stdout.

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

def __init__(self, command: List[str]) -> None:
    self._command = command
    self._process: Optional[subprocess.Popen[str]] = None
    self._start()

Functions

send

send(request: MCPRequest) -> MCPResponse

Write request as JSON line, read response line.

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

def send(self, request: MCPRequest) -> MCPResponse:
    """Write request as JSON line, read response line."""
    proc = self._process
    if proc is None or proc.stdin is None or proc.stdout is None:
        raise RuntimeError("Transport process is not running")

    line = request.to_json() + "\n"
    proc.stdin.write(line)
    proc.stdin.flush()

    response_line = proc.stdout.readline()
    if not response_line:
        raise RuntimeError("No response from subprocess")
    return MCPResponse.from_json(response_line.strip())

close

close() -> None

Terminate the subprocess.

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

def close(self) -> None:
    """Terminate the subprocess."""
    if self._process is not None:
        self._process.terminate()
        self._process.wait(timeout=5)
        self._process = None