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.

Sends the required client info and protocol version, then confirms with a notifications/initialized notification as required by the MCP specification.

Returns the server capabilities.

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

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

    Sends the required client info and protocol version, then
    confirms with a ``notifications/initialized`` notification
    as required by the MCP specification.

    Returns the server capabilities.
    """
    params = {
        "protocolVersion": "2025-03-26",
        "capabilities": {},
        "clientInfo": {"name": "openjarvis", "version": "0.1.0"},
    }
    response = self._send("initialize", params)
    self._initialized = True
    self._capabilities = response.result.get("capabilities", {})
    # Send the required initialized notification per MCP spec
    self.notify("notifications/initialized")
    return response.result

notify

notify(method: str, params: Dict[str, Any] | None = None) -> None

Send a JSON-RPC notification (no response expected).

Per JSON-RPC 2.0 spec, notifications omit the id field entirely.

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

def notify(self, method: str, params: Dict[str, Any] | None = None) -> None:
    """Send a JSON-RPC notification (no response expected).

    Per JSON-RPC 2.0 spec, notifications omit the ``id`` field entirely.
    """
    request = MCPRequest(
        method=method,
        params=params or {},
        id=None,  # None → no id field in JSON (notification)
    )
    self._transport.send_notification(request)

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: Optional[int | str] = 0, jsonrpc: str = '2.0')

JSON-RPC 2.0 request message.

Set id to None to create a JSON-RPC notification (no id field will appear in the serialized output, and no response is expected).

Functions

to_dict

to_dict() -> Dict[str, Any]

Return a dict suitable for JSON serialization.

Omits the id key when it is None (notification).

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

def to_dict(self) -> Dict[str, Any]:
    """Return a dict suitable for JSON serialization.

    Omits the ``id`` key when it is ``None`` (notification).
    """
    obj: Dict[str, Any] = {
        "jsonrpc": self.jsonrpc,
        "method": self.method,
        "params": self.params,
    }
    if self.id is not None:
        obj["id"] = self.id
    return obj

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(self.to_dict())

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."""

send_notification

send_notification(request: MCPRequest) -> None

Send a JSON-RPC notification (no response expected).

The default implementation delegates to :meth:send and discards the response. Transports may override this when the server returns no body for notifications (e.g. HTTP 202 Accepted).

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

def send_notification(self, request: MCPRequest) -> None:
    """Send a JSON-RPC notification (no response expected).

    The default implementation delegates to :meth:`send` and discards the
    response.  Transports may override this when the server returns no
    body for notifications (e.g. HTTP 202 Accepted).
    """
    self.send(request)

close abstractmethod

close() -> None

Release transport resources.

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

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

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

StreamableHTTPTransport

StreamableHTTPTransport(url: str, *, connect_timeout: float = 10.0, request_timeout: float = 60.0)

Bases: MCPTransport

MCP Streamable HTTP transport (JSON-RPC over HTTP).

Uses a persistent httpx.Client session, tracks the Mcp-Session-Id header, and sends the Accept header required by the MCP Streamable HTTP specification.

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

def __init__(
    self,
    url: str,
    *,
    connect_timeout: float = 10.0,
    request_timeout: float = 60.0,
) -> None:
    import httpx

    self._url = url
    self._session_id: Optional[str] = None
    self._client = httpx.Client(
        timeout=httpx.Timeout(
            connect=connect_timeout,
            read=request_timeout,
            write=request_timeout,
            pool=connect_timeout,
        ),
    )

Functions

send

send(request: MCPRequest) -> MCPResponse

Send request via HTTP POST following the MCP Streamable HTTP spec.

Handles both application/json and text/event-stream responses as allowed by the MCP Streamable HTTP specification.

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

def send(self, request: MCPRequest) -> MCPResponse:
    """Send request via HTTP POST following the MCP Streamable HTTP spec.

    Handles both ``application/json`` and ``text/event-stream`` responses
    as allowed by the MCP Streamable HTTP specification.
    """
    response = self._post(request)
    content_type = response.headers.get("content-type", "")
    body = response.text
    if "text/event-stream" in content_type or body.lstrip().startswith("event:"):
        body = self._extract_json_from_sse(body)
    return MCPResponse.from_json(body)

send_notification

send_notification(request: MCPRequest) -> None

Send a notification — accept any 2xx, don't parse the body.

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

def send_notification(self, request: MCPRequest) -> None:
    """Send a notification — accept any 2xx, don't parse the body."""
    # Track session id but don't try to parse a JSON-RPC response.
    # Servers may return 202 Accepted with an empty body.
    self._post(request)

close

close() -> None

Close the underlying httpx client.

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

def close(self) -> None:
    """Close the underlying httpx client."""
    self._client.close()