Skip to content

MCP

pycodeloop.core.mcp.MCPServer dataclass

Import

You can import the MCPServer class with:

from pycodeloop.core.mcp import MCPServer, load_mcp_tools
Example

class pycodeloop.core.mcp.MCPServer

server = MCPServer(
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem", "."],
)
tools = load_mcp_tools(server)

config = Config(tools=DEFAULT_TOOLS + tools)

Parameters:

Name Type Description Default
command str

Executable that speaks MCP over stdio.

 required
args Optional[List[str]]

Arguments passed to the command.

 list()
env Optional[Dict[str, str]]

Extra environment variables.

 None
Source code in pycodeloop/core/mcp.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@dataclass
class MCPServer:
    """
    Import:
        You can import the **MCPServer** class with:

            from pycodeloop.core.mcp import MCPServer, load_mcp_tools

    Example:
        `class` pycodeloop.core.mcp.MCPServer

            server = MCPServer(
                command="npx",
                args=["-y", "@modelcontextprotocol/server-filesystem", "."],
            )
            tools = load_mcp_tools(server)

            config = Config(tools=DEFAULT_TOOLS + tools)

    Args:
        command (str): Executable that speaks MCP over stdio.
        args (Optional[List[str]]): Arguments passed to the command.
        env (Optional[Dict[str, str]]): Extra environment variables.
    """

    command: str
    args: list[str] = field(default_factory=list)
    env: dict[str, str] | None = None

args = field(default_factory=list) class-attribute instance-attribute

command instance-attribute

env = None class-attribute instance-attribute

__init__(command, args=list(), env=None)

pycodeloop.core.mcp.MCPClient

One MCP server subprocess on a dedicated background event loop — bridges its async session into synchronous Tool.run() calls.

Source code in pycodeloop/core/mcp.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class MCPClient:
    """One MCP server subprocess on a dedicated background event loop —
    bridges its async session into synchronous `Tool.run()` calls."""

    def __init__(self, server: MCPServer) -> None:
        self.server = server
        self._loop = asyncio.new_event_loop()
        self._thread = threading.Thread(target=self._loop.run_forever, daemon=True)
        self._thread.start()
        self._exit_stack = None
        self._session = None

        self._run(self._connect())

    def _run(self, coro, timeout: float = 30):
        future = asyncio.run_coroutine_threadsafe(coro, self._loop)
        return future.result(timeout=timeout)

    async def _connect(self) -> None:
        from mcp import ClientSession, StdioServerParameters
        from mcp.client.stdio import stdio_client

        self._exit_stack = AsyncExitStack()
        params = StdioServerParameters(
            command=self.server.command,
            args=self.server.args,
            env=self.server.env,
        )

        read, write = await self._exit_stack.enter_async_context(stdio_client(params))
        session = await self._exit_stack.enter_async_context(ClientSession(read, write))
        await session.initialize()

        self._session = session

    def list_tools(self) -> list[dict]:
        return self._run(self._list_tools())

    async def _list_tools(self) -> list[dict]:
        result = await self._session.list_tools()
        return [
            {
                "name": tool.name,
                "description": tool.description or "",
                "input_schema": (
                    getattr(tool, "input_schema", None)
                    or getattr(tool, "inputSchema", None)
                    or {"type": "object", "properties": {}}
                ),
            }
            for tool in result.tools
        ]

    def call_tool(self, name: str, arguments: dict) -> str:
        return self._run(self._call_tool(name, arguments), timeout=120)

    async def _call_tool(self, name: str, arguments: dict) -> str:
        result = await self._session.call_tool(name, arguments)
        parts = [block.text for block in result.content if getattr(block, "text", None)]
        return "\n".join(parts) if parts else str(result.content)

    def close(self) -> None:
        if self._exit_stack is not None:
            self._run(self._exit_stack.aclose())

        self._loop.call_soon_threadsafe(self._loop.stop)
        self._thread.join(timeout=5)

server = server instance-attribute

__init__(server)

Source code in pycodeloop/core/mcp.py 
52
53
54
55
56
57
58
59
60
def __init__(self, server: MCPServer) -> None:
    self.server = server
    self._loop = asyncio.new_event_loop()
    self._thread = threading.Thread(target=self._loop.run_forever, daemon=True)
    self._thread.start()
    self._exit_stack = None
    self._session = None

    self._run(self._connect())

call_tool(name, arguments)

Source code in pycodeloop/core/mcp.py 
101
102
def call_tool(self, name: str, arguments: dict) -> str:
    return self._run(self._call_tool(name, arguments), timeout=120)

close()

Source code in pycodeloop/core/mcp.py 
109
110
111
112
113
114
def close(self) -> None:
    if self._exit_stack is not None:
        self._run(self._exit_stack.aclose())

    self._loop.call_soon_threadsafe(self._loop.stop)
    self._thread.join(timeout=5)

list_tools()

Source code in pycodeloop/core/mcp.py 
83
84
def list_tools(self) -> list[dict]:
    return self._run(self._list_tools())

pycodeloop.core.mcp.MCPTool

Bases: Tool

Adapts one remote MCP tool into the pycodeloop Tool ABC.

Source code in pycodeloop/core/mcp.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
class MCPTool(Tool):
    """Adapts one remote MCP tool into the pycodeloop Tool ABC."""

    dangerous = True

    def __init__(self, client: MCPClient, schema: dict) -> None:
        self.client = client
        self.name = schema["name"]
        self.description = schema["description"]
        self.parameters = schema["input_schema"]

    def preview(self, **kwargs) -> str:
        args = ", ".join(f"{key}={value!r}" for key, value in kwargs.items())
        return f"{self.name}({args})"

    def run(self, **kwargs) -> ToolResult:
        try:
            output = self.client.call_tool(self.name, kwargs)
        except Exception as exc:
            return ToolResult(
                output=f"Error calling MCP tool '{self.name}': {exc}",
                is_error=True,
            )

        return ToolResult(output=output)

client = client instance-attribute

dangerous = True class-attribute instance-attribute

description = schema['description'] instance-attribute

name = schema['name'] instance-attribute

parameters = schema['input_schema'] instance-attribute

__init__(client, schema)

Source code in pycodeloop/core/mcp.py 
122
123
124
125
126
def __init__(self, client: MCPClient, schema: dict) -> None:
    self.client = client
    self.name = schema["name"]
    self.description = schema["description"]
    self.parameters = schema["input_schema"]

preview(**kwargs)

Source code in pycodeloop/core/mcp.py 
128
129
130
def preview(self, **kwargs) -> str:
    args = ", ".join(f"{key}={value!r}" for key, value in kwargs.items())
    return f"{self.name}({args})"

run(**kwargs)

Source code in pycodeloop/core/mcp.py 
132
133
134
135
136
137
138
139
140
141
def run(self, **kwargs) -> ToolResult:
    try:
        output = self.client.call_tool(self.name, kwargs)
    except Exception as exc:
        return ToolResult(
            output=f"Error calling MCP tool '{self.name}': {exc}",
            is_error=True,
        )

    return ToolResult(output=output)

pycodeloop.core.mcp.load_mcp_tools(server)

Connect to an MCP server, return its tools as pycodeloop Tools.

The client's subprocess/thread are registered for cleanup at interpreter exit — nothing else in the app ever called close(), which otherwise leaked a live MCP server subprocess per connection.

Source code in pycodeloop/core/mcp.py 
144
145
146
147
148
149
150
151
152
153
def load_mcp_tools(server: MCPServer) -> list[Tool]:
    """Connect to an MCP server, return its tools as pycodeloop Tools.

    The client's subprocess/thread are registered for cleanup at
    interpreter exit — nothing else in the app ever called `close()`,
    which otherwise leaked a live MCP server subprocess per connection."""
    client = MCPClient(server)
    atexit.register(client.close)

    return [MCPTool(client, schema) for schema in client.list_tools()]