feat: Initial commit

pyproject.toml

@@ -0,0 +1,18 @@
+[build-system]
+requires = ["setuptools>=68"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "mcp-api"
+version = "0.1.0"
+requires-python = ">=3.10"
+dependencies = ["mcp>=1.0"]
+
+[project.optional-dependencies]
+dev = ["pytest", "pytest-asyncio", "anyio"]
+
+[tool.setuptools.packages.find]
+where = ["src"]
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"

src/mcp_api/__init__.py

@@ -0,0 +1,5 @@
+from .config import ServerConfig, from_dict, load_claude_json
+from .tool import MCPTool
+from .toolbox import MCPToolbox
+
+__all__ = ["MCPToolbox", "MCPTool", "ServerConfig", "from_dict", "load_claude_json"]

src/mcp_api/config.py

@@ -0,0 +1,36 @@
+"""Config loading from claude.json or plain dicts."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from pathlib import Path
+
+
+@dataclass
+class ServerConfig:
+    name: str
+    command: str
+    args: list[str] = field(default_factory=list)
+    env: dict[str, str] | None = None
+
+
+def load_claude_json(path: Path | None = None) -> dict[str, ServerConfig]:
+    """Load mcpServers from ~/.claude.json (or a custom path)."""
+    if path is None:
+        path = Path.home() / ".claude.json"
+    data = json.loads(path.read_text())
+    return from_dict(data.get("mcpServers", {}))
+
+
+def from_dict(servers: dict) -> dict[str, ServerConfig]:
+    """Parse a dict in the claude.json mcpServers format."""
+    result: dict[str, ServerConfig] = {}
+    for name, cfg in servers.items():
+        result[name] = ServerConfig(
+            name=name,
+            command=cfg["command"],
+            args=cfg.get("args", []),
+            env=cfg.get("env"),
+        )
+    return result

src/mcp_api/session.py

@@ -0,0 +1,59 @@
+"""Lifecycle management for a single MCP server connection."""
+
+from __future__ import annotations
+
+from contextlib import AsyncExitStack
+
+from mcp import ClientSession, StdioServerParameters
+from mcp import types
+from mcp.client.stdio import stdio_client
+
+from .config import ServerConfig
+
+
+class ServerSession:
+    """Manages one MCP server process and its ClientSession."""
+
+    def __init__(self, config: ServerConfig) -> None:
+        self._config = config
+        self._stack = AsyncExitStack()
+        self._session: ClientSession | None = None
+
+    async def connect(self) -> None:
+        params = StdioServerParameters(
+            command=self._config.command,
+            args=self._config.args,
+            env=self._config.env,
+        )
+        read, write = await self._stack.enter_async_context(stdio_client(params))
+        self._session = await self._stack.enter_async_context(
+            ClientSession(read, write)
+        )
+        await self._session.initialize()
+
+    async def disconnect(self) -> None:
+        await self._stack.aclose()
+        self._session = None
+
+    async def list_tools(self) -> list[types.Tool]:
+        assert self._session is not None, "Not connected"
+        result = await self._session.list_tools()
+        return result.tools
+
+    async def call_tool(self, name: str, arguments: dict) -> str:
+        assert self._session is not None, "Not connected"
+        result = await self._session.call_tool(name, arguments)
+        if result.isError:
+            error_text = _extract_text(result.content)
+            raise RuntimeError(f"Tool '{name}' returned an error: {error_text}")
+        return _extract_text(result.content)
+
+
+def _extract_text(content: list) -> str:
+    parts: list[str] = []
+    for item in content:
+        if hasattr(item, "text"):
+            parts.append(item.text)
+        else:
+            parts.append(str(item))
+    return "\n".join(parts)

src/mcp_api/tool.py

@@ -0,0 +1,31 @@
+"""Callable wrapper around a single MCP tool."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from .session import ServerSession
+
+
+class MCPTool:
+    """A Python callable that invokes an MCP tool on a remote server."""
+
+    def __init__(self, name: str, description: str, input_schema: dict, session: ServerSession) -> None:
+        self.name = name
+        self.description = description
+        self.input_schema = input_schema
+        self._session = session
+
+    async def __call__(self, **kwargs: Any) -> str:
+        return await self._session.call_tool(self.name, kwargs)
+
+    def as_anthropic_tool(self) -> dict:
+        return {
+            "name": self.name,
+            "description": self.description,
+            "input_schema": self.input_schema,
+        }
+
+    def __repr__(self) -> str:
+        return f"MCPTool(name={self.name!r})"

src/mcp_api/toolbox.py

@@ -0,0 +1,141 @@
+"""MCPToolbox: main entry point for the library."""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+from typing import Any
+
+from .config import ServerConfig, from_dict, load_claude_json
+from .session import ServerSession
+from .tool import MCPTool
+
+
+class _ServerNamespace:
+    """Attribute-style access to tools belonging to one server.
+
+    Usage: ``toolbox.pin.list_plugins(output="/tmp/out.txt")``
+    """
+
+    def __init__(self, session: ServerSession, all_tools: dict[str, MCPTool]) -> None:
+        self._session = session
+        self._tools = {
+            name: tool for name, tool in all_tools.items() if tool._session is session
+        }
+
+    def __getattr__(self, name: str) -> MCPTool:
+        if name in self._tools:
+            return self._tools[name]
+        raise AttributeError(name)
+
+    def __dir__(self) -> list[str]:
+        return list(self._tools)
+
+    def __repr__(self) -> str:
+        return f"_ServerNamespace(tools={list(self._tools)})"
+
+
+class MCPToolbox:
+    """Connects to one or more MCP servers and exposes their tools."""
+
+    def __init__(self, configs: dict[str, ServerConfig]) -> None:
+        self._configs = configs
+        self._sessions: dict[str, ServerSession] = {}
+        self._tools: dict[str, MCPTool] = {}
+
+    # ------------------------------------------------------------------ #
+    # Constructors                                                         #
+    # ------------------------------------------------------------------ #
+
+    @classmethod
+    def from_claude_json(cls, path: Path | None = None) -> MCPToolbox:
+        """Load server config from ~/.claude.json (or a custom path)."""
+        return cls(load_claude_json(path))
+
+    @classmethod
+    def from_config(cls, servers: dict) -> MCPToolbox:
+        """Load server config from a plain dict (claude.json mcpServers format)."""
+        return cls(from_dict(servers))
+
+    # ------------------------------------------------------------------ #
+    # Async context manager                                                #
+    # ------------------------------------------------------------------ #
+
+    async def __aenter__(self) -> MCPToolbox:
+        await self.discover()
+        return self
+
+    async def __aexit__(self, *_: Any) -> None:
+        await asyncio.gather(
+            *(s.disconnect() for s in self._sessions.values()),
+            return_exceptions=True,
+        )
+
+    # ------------------------------------------------------------------ #
+    # Discovery                                                            #
+    # ------------------------------------------------------------------ #
+
+    async def discover(self) -> None:
+        """Connect to all servers in parallel and discover their tools."""
+        await asyncio.gather(*(self._connect_server(cfg) for cfg in self._configs.values()))
+
+    async def _connect_server(self, cfg: ServerConfig) -> None:
+        session = ServerSession(cfg)
+        await session.connect()
+        self._sessions[cfg.name] = session
+        for mcp_tool in await session.list_tools():
+            tool = MCPTool(
+                name=mcp_tool.name,
+                description=mcp_tool.description or "",
+                input_schema=mcp_tool.inputSchema if isinstance(mcp_tool.inputSchema, dict)
+                             else mcp_tool.inputSchema.model_dump(),
+                session=session,
+            )
+            self._tools[tool.name] = tool
+
+    # ------------------------------------------------------------------ #
+    # Tool access                                                          #
+    # ------------------------------------------------------------------ #
+
+    @property
+    def tools(self) -> list[MCPTool]:
+        return list(self._tools.values())
+
+    def __getattr__(self, name: str) -> _ServerNamespace | MCPTool:
+        # toolbox.pin  → ServerNamespace for that server
+        if name in self._sessions:
+            return _ServerNamespace(self._sessions[name], self._tools)
+        # toolbox.list_plugins  → MCPTool directly (flat access)
+        if name in self._tools:
+            return self._tools[name]
+        raise AttributeError(name)
+
+    def __getitem__(self, name: str) -> MCPTool:
+        return self._tools[name]
+
+    # ------------------------------------------------------------------ #
+    # LLM integration helpers                                              #
+    # ------------------------------------------------------------------ #
+
+    def as_anthropic_tools(self) -> list[dict]:
+        """Return all tools in Anthropic API format."""
+        return [t.as_anthropic_tool() for t in self._tools.values()]
+
+    async def call_tool(self, name: str, arguments: dict | None = None) -> str:
+        """Call a tool by name and return its text output."""
+        return await self._tools[name]._session.call_tool(name, arguments or {})
+
+    async def handle_anthropic_tool_use(self, tool_use_block: Any) -> dict:
+        """Dispatch an Anthropic ToolUseBlock and return a tool_result dict."""
+        try:
+            content = await self.call_tool(tool_use_block.name, tool_use_block.input)
+            is_error = False
+        except Exception as exc:
+            content = str(exc)
+            is_error = True
+        return {
+            "type": "tool_result",
+            "tool_use_id": tool_use_block.id,
+            "content": content,
+            "is_error": is_error,
+        }

tests/conftest.py

@@ -0,0 +1,33 @@
+"""Shared pytest fixtures."""
+
+import sys
+from pathlib import Path
+
+import pytest
+import pytest_asyncio
+
+from mcp_api import MCPToolbox
+
+FIXTURES = Path(__file__).parent / "fixtures"
+
+
+@pytest.fixture
+def server_config() -> dict:
+    """Config dict for the two test MCP servers, using the current Python interpreter."""
+    python = sys.executable
+    return {
+        "echo": {
+            "command": python,
+            "args": [str(FIXTURES / "echo_server.py")],
+        },
+        "math": {
+            "command": python,
+            "args": [str(FIXTURES / "math_server.py")],
+        },
+    }
+
+
+@pytest_asyncio.fixture
+async def toolbox(server_config: dict) -> MCPToolbox:
+    async with MCPToolbox.from_config(server_config) as tb:
+        yield tb

tests/fixtures/echo_server.py

@@ -0,0 +1,21 @@
+"""Simple MCP server used for testing: string tools."""
+
+from mcp.server.fastmcp import FastMCP
+
+mcp = FastMCP("echo")
+
+
+@mcp.tool()
+def echo(message: str) -> str:
+    """Return the message unchanged."""
+    return message
+
+
+@mcp.tool()
+def reverse(text: str) -> str:
+    """Return the text reversed."""
+    return text[::-1]
+
+
+if __name__ == "__main__":
+    mcp.run("stdio")

tests/fixtures/math_server.py

@@ -0,0 +1,21 @@
+"""Simple MCP server used for testing: numeric tools."""
+
+from mcp.server.fastmcp import FastMCP
+
+mcp = FastMCP("math")
+
+
+@mcp.tool()
+def add(a: int, b: int) -> int:
+    """Return the sum of a and b."""
+    return a + b
+
+
+@mcp.tool()
+def multiply(a: int, b: int) -> int:
+    """Return the product of a and b."""
+    return a * b
+
+
+if __name__ == "__main__":
+    mcp.run("stdio")

tests/fixtures/test_claude.json

@@ -0,0 +1,16 @@
+{
+  "mcpServers": {
+    "echo": {
+      "command": "/home/matteo/Gitted/Thesis/python-mcp-api/.venv/bin/python",
+      "args": [
+        "/home/matteo/Gitted/Thesis/python-mcp-api/tests/fixtures/echo_server.py"
+      ]
+    },
+    "math": {
+      "command": "/home/matteo/Gitted/Thesis/python-mcp-api/.venv/bin/python",
+      "args": [
+        "/home/matteo/Gitted/Thesis/python-mcp-api/tests/fixtures/math_server.py"
+      ]
+    }
+  }
+}

tests/test_toolbox.py

@@ -0,0 +1,141 @@
+"""Tests for MCPToolbox."""
+
+import pytest
+
+from mcp_api import MCPTool, MCPToolbox
+
+
+# ---------------------------------------------------------------------------
+# Discovery
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_discovers_all_tools(toolbox: MCPToolbox) -> None:
+    names = {t.name for t in toolbox.tools}
+    assert names == {"echo", "reverse", "add", "multiply"}
+
+
+@pytest.mark.asyncio
+async def test_tool_has_description(toolbox: MCPToolbox) -> None:
+    for tool in toolbox.tools:
+        assert tool.description, f"{tool.name} has no description"
+
+
+@pytest.mark.asyncio
+async def test_tool_has_input_schema(toolbox: MCPToolbox) -> None:
+    for tool in toolbox.tools:
+        assert isinstance(tool.input_schema, dict)
+        assert tool.input_schema.get("type") == "object"
+
+
+# ---------------------------------------------------------------------------
+# Tool access
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_getitem(toolbox: MCPToolbox) -> None:
+    assert isinstance(toolbox["echo"], MCPTool)
+
+
+@pytest.mark.asyncio
+async def test_flat_attribute_access(toolbox: MCPToolbox) -> None:
+    # "echo" is also a server name → _ServerNamespace takes priority
+    # Use a tool name that doesn't collide with a server name for flat access
+    assert isinstance(toolbox.reverse, MCPTool)
+
+
+@pytest.mark.asyncio
+async def test_server_namespace_attribute_access(toolbox: MCPToolbox) -> None:
+    assert isinstance(toolbox.echo.echo, MCPTool)
+    assert isinstance(toolbox.math.add, MCPTool)
+
+
+@pytest.mark.asyncio
+async def test_unknown_attribute_raises(toolbox: MCPToolbox) -> None:
+    with pytest.raises(AttributeError):
+        _ = toolbox.nonexistent
+
+
+# ---------------------------------------------------------------------------
+# Tool calls — direct
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_call_tool_echo(toolbox: MCPToolbox) -> None:
+    result = await toolbox.call_tool("echo", {"message": "hello"})
+    assert result == "hello"
+
+
+@pytest.mark.asyncio
+async def test_call_tool_reverse(toolbox: MCPToolbox) -> None:
+    result = await toolbox.call_tool("reverse", {"text": "abc"})
+    assert result == "cba"
+
+
+@pytest.mark.asyncio
+async def test_call_tool_add(toolbox: MCPToolbox) -> None:
+    result = await toolbox.call_tool("add", {"a": 3, "b": 4})
+    assert result == "7"
+
+
+@pytest.mark.asyncio
+async def test_call_tool_multiply(toolbox: MCPToolbox) -> None:
+    result = await toolbox.call_tool("multiply", {"a": 6, "b": 7})
+    assert result == "42"
+
+
+# ---------------------------------------------------------------------------
+# Tool calls — via callable / attribute access
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_callable_tool(toolbox: MCPToolbox) -> None:
+    result = await toolbox["echo"](message="world")
+    assert result == "world"
+
+
+@pytest.mark.asyncio
+async def test_server_namespace_call(toolbox: MCPToolbox) -> None:
+    result = await toolbox.math.add(a=10, b=5)
+    assert result == "15"
+
+
+# ---------------------------------------------------------------------------
+# Anthropic format
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_as_anthropic_tools_structure(toolbox: MCPToolbox) -> None:
+    defs = toolbox.as_anthropic_tools()
+    assert len(defs) == 4
+    for d in defs:
+        assert {"name", "description", "input_schema"} == d.keys()
+        assert isinstance(d["input_schema"], dict)
+
+
+@pytest.mark.asyncio
+async def test_handle_anthropic_tool_use_success(toolbox: MCPToolbox) -> None:
+    class FakeBlock:
+        type = "tool_use"
+        id = "tu_123"
+        name = "echo"
+        input = {"message": "ping"}
+
+    result = await toolbox.handle_anthropic_tool_use(FakeBlock())
+    assert result["type"] == "tool_result"
+    assert result["tool_use_id"] == "tu_123"
+    assert result["content"] == "ping"
+    assert result["is_error"] is False
+
+
+@pytest.mark.asyncio
+async def test_handle_anthropic_tool_use_error(toolbox: MCPToolbox) -> None:
+    class FakeBlock:
+        type = "tool_use"
+        id = "tu_456"
+        name = "nonexistent_tool"
+        input = {}
+
+    result = await toolbox.handle_anthropic_tool_use(FakeBlock())
+    assert result["is_error"] is True
+    assert result["tool_use_id"] == "tu_456"