tmcp-runner 0.4.0

Python library for embedding MCP (Model Context Protocol) capabilities into applications, like Claude Desktop and Cursor do internally. Sponsored by TowardsAGI.

🧩 tmcp_runner

A Python library for embedding Model Context Protocol (MCP) capabilities into your applications, just like Claude Desktop and Cursor do internally.

Overview

tmcp_runner enables your Python applications to:

• 🔌 Connect to any MCP-compliant server using the official Anthropic MCP SDK
• 🔍 Discover available tools, resources, and prompts from MCP servers
• ⚡ Execute MCP tools programmatically with full async support
• 🌐 Work with multiple MCP servers simultaneously
• 🎯 Use the same configuration format as Claude Desktop

Installation

pip install tmcp-runner

Or install from source:

git clone https://github.com/QuantumicsAI/tmcp_runner.git
cd tmcp_runner
pip install -e .

Quick Start

1. Create MCP Configuration

Create an mcp_config.json file with your MCP servers (same format as Claude Desktop):

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/directory"],
      "transport": "stdio"
    },
    "postgres": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres"],
      "env": {
        "POSTGRES_CONNECTION_STRING": "postgresql://user:pass@localhost:5432/dbname"
      },
      "transport": "stdio"
    }
  }
}

2. Import and Use in Your Application

import asyncio
from tmcp_runner import TMCPRunner

async def main():
    # Initialize the MCP runner
    runner = TMCPRunner("mcp_config.json")
    
    # Connect to all configured MCP servers
    await runner.connect_all()
    
    # Discover available tools
    discovery = await runner.discover_all()
    print(f"Connected to {len(discovery)} MCP servers")
    
    # Execute a tool
    result = await runner.execute_tool(
        server_name="filesystem",
        tool_name="read_file",
        arguments={"path": "/path/to/file.txt"}
    )
    
    # Process results
    for content in result:
        if hasattr(content, 'text'):
            print(content.text)
    
    # Cleanup when done
    await runner.disconnect_all()

asyncio.run(main())

✅ Tested Agent Integration Examples

We provide 3 complete, tested examples showing how to integrate tmcp_runner with different AI agent frameworks:

1. Simple Agent (test_simple_agent.py)

Basic pattern for custom AI agents. 4/4 tests passed ✅

• Tool discovery and management
• Query handling
• Direct tool execution
• Perfect for prototyping and learning

2. LangChain Agent (test_langchain_agent.py)

Integration with LangChain framework. 6/6 tests passed ✅

• MCP tools wrapped as LangChain-compatible tools
• AgentExecutor integration
• Tool management for LangChain
• Perfect for LangChain projects

3. Pydantic Agent (test_pydantic_agent.py)

Type-safe agent with Pydantic validation. 7/7 tests passed ✅

• Type-safe tool definitions
• Validated execution results
• Runtime type checking
• Perfect for production systems

Run the tests:

python3 tests/test_simple_agent.py
python3 tests/test_langchain_agent.py
python3 tests/test_pydantic_agent.py

See tests/README_AGENT_TESTS.md for complete documentation.

Core API

TMCPRunner

Main class for managing multiple MCP server connections in your application.

from tmcp_runner import TMCPRunner

runner = TMCPRunner("path/to/config.json")

Key Methods

# Connection Management
await runner.connect_all()                          # Connect to all servers
await runner.connect_server("server-name")          # Connect to specific server
await runner.disconnect_all()                       # Clean up all connections

# Discovery
servers = runner.list_servers()                     # List configured servers
discovery = await runner.discover_all()             # Discover all tools & resources

# Tool Execution
result = await runner.execute_tool(
    server_name="filesystem",
    tool_name="read_file",
    arguments={"path": "/file.txt"}
)

# Resource Reading
content = await runner.read_resource(
    server_name="github",
    uri="repo://owner/repo/README.md"
)

MCPClient

For direct interaction with a single MCP server:

from tmcp_runner import MCPClient

# Configure a server
config = {
    "command": "npx",
    "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
    "transport": "stdio"
}

# Use with context manager (auto-cleanup)
async with MCPClient("filesystem", config) as client:
    # List available tools
    tools = await client.list_tools()
    
    # Execute a tool
    result = await client.call_tool("read_file", {"path": "/tmp/test.txt"})
    
    # List resources
    resources = await client.list_resources()

Integration Examples

Example 1: Adding MCP to a Web Application

from fastapi import FastAPI
from tmcp_runner import TMCPRunner

app = FastAPI()
runner = TMCPRunner("mcp_config.json")

@app.on_event("startup")
async def startup():
    """Initialize MCP connections when app starts"""
    await runner.connect_all()
    print(f"✅ MCP servers connected: {runner.list_servers()}")

@app.on_event("shutdown")
async def shutdown():
    """Cleanup MCP connections when app shuts down"""
    await runner.disconnect_all()

@app.post("/execute-tool")
async def execute_tool(server: str, tool: str, arguments: dict):
    """Expose MCP tool execution via API"""
    result = await runner.execute_tool(server, tool, arguments)
    return {"result": [{"text": c.text} for c in result if hasattr(c, 'text')]}

Example 2: AI Agent with MCP Tools

from tmcp_runner import TMCPRunner
import asyncio

class AIAgent:
    def __init__(self, mcp_config_path: str):
        self.mcp = TMCPRunner(mcp_config_path)
        self.available_tools = {}
    
    async def initialize(self):
        """Setup MCP connections and discover tools"""
        await self.mcp.connect_all()
        
        # Discover all available tools
        discovery = await self.mcp.discover_all()
        for server_name, info in discovery.items():
            for tool in info.get('tools', []):
                tool_id = f"{server_name}.{tool['name']}"
                self.available_tools[tool_id] = {
                    'server': server_name,
                    'name': tool['name'],
                    'description': tool['description']
                }
    
    async def use_tool(self, tool_id: str, arguments: dict):
        """Execute an MCP tool"""
        tool = self.available_tools[tool_id]
        result = await self.mcp.execute_tool(
            tool['server'],
            tool['name'],
            arguments
        )
        return result
    
    async def cleanup(self):
        """Cleanup MCP connections"""
        await self.mcp.disconnect_all()

# Usage
async def main():
    agent = AIAgent("mcp_config.json")
    await agent.initialize()
    
    # Agent can now use any configured MCP tool
    result = await agent.use_tool(
        "filesystem.read_file",
        {"path": "/data/context.txt"}
    )
    
    await agent.cleanup()

asyncio.run(main())

Example 3: Background Task Processing

from tmcp_runner import TMCPRunner
import asyncio
from typing import List, Dict

class MCPTaskProcessor:
    def __init__(self, config_path: str):
        self.runner = TMCPRunner(config_path)
        self.task_queue = asyncio.Queue()
    
    async def start(self):
        """Start the processor"""
        await self.runner.connect_all()
        asyncio.create_task(self._process_tasks())
    
    async def _process_tasks(self):
        """Process tasks from queue"""
        while True:
            task = await self.task_queue.get()
            try:
                result = await self.runner.execute_tool(
                    task['server'],
                    task['tool'],
                    task['arguments']
                )
                task['callback'](result)
            except Exception as e:
                task['error_callback'](e)
            finally:
                self.task_queue.task_done()
    
    async def submit_task(self, server: str, tool: str, arguments: dict, callback, error_callback):
        """Submit a task for processing"""
        await self.task_queue.put({
            'server': server,
            'tool': tool,
            'arguments': arguments,
            'callback': callback,
            'error_callback': error_callback
        })
    
    async def stop(self):
        """Stop the processor"""
        await self.task_queue.join()
        await self.runner.disconnect_all()

Configuration

Stdio Transport (Local Servers)

For MCP servers that run as local processes:

{
  "server-name": {
    "command": "npx",
    "args": ["-y", "@modelcontextprotocol/server-name", "arg1"],
    "env": {
      "API_KEY": "your-key"
    },
    "transport": "stdio"
  }
}

SSE Transport (Remote Servers)

For remote MCP servers using Server-Sent Events:

{
  "remote-server": {
    "url": "https://your-mcp-server.com/sse",
    "transport": "sse"
  }
}

Supported MCP Servers

Works with any MCP-compliant server, including:

Official Anthropic Servers

• filesystem - File system operations
• github - GitHub API integration
• postgres - PostgreSQL database access
• sqlite - SQLite database access
• puppeteer - Web automation
• gdrive - Google Drive integration
• slack - Slack integration
• And many more...

Browse available servers at mcpservers.org

Architecture

┌─────────────────────────┐
│   Your Application      │
│  (Web App, AI Agent,    │
│   Background Service)   │
└───────────┬─────────────┘
            │
            │ imports
            │
    ┌───────▼────────┐
    │  TMCPRunner    │
    │  (Library)     │
    └───────┬────────┘
            │
            │ manages
            │
    ┌───────▼────────┐
    │   MCPClient    │
    │   (per server) │
    └───────┬────────┘
            │
            │ uses
            │
    ┌───────▼────────┐
    │  Official MCP  │
    │   Python SDK   │
    └───────┬────────┘
            │
    ┌───────▼────────┐
    │  MCP Servers   │
    │ (Tools, Data)  │
    └────────────────┘

How It Works (Like Claude Desktop)

1. Configuration Loading: Reads MCP server configs from JSON file
2. Server Connection: Spawns stdio processes or connects to SSE endpoints
3. Discovery: Queries each server for available tools, resources, and prompts
4. Tool Execution: Executes tools via JSON-RPC and returns typed results
5. Resource Reading: Fetches resource contents by URI

Advanced Usage

Concurrent Tool Execution

async def execute_multiple_tools():
    runner = TMCPRunner("mcp_config.json")
    await runner.connect_all()
    
    # Execute multiple tools concurrently
    results = await asyncio.gather(
        runner.execute_tool("server1", "tool1", {"arg": "val1"}),
        runner.execute_tool("server2", "tool2", {"arg": "val2"}),
        runner.execute_tool("server3", "tool3", {"arg": "val3"}),
    )
    
    await runner.disconnect_all()
    return results

Dynamic Server Management

async def dynamic_servers():
    runner = TMCPRunner("mcp_config.json")
    
    # Connect only to specific servers
    await runner.connect_server("filesystem")
    await runner.connect_server("postgres")
    
    # Later, connect to more servers
    await runner.connect_server("github")
    
    # Disconnect specific server
    await runner.disconnect_server("filesystem")
    
    await runner.disconnect_all()

Error Handling

async def robust_execution():
    runner = TMCPRunner("mcp_config.json")
    
    try:
        await runner.connect_all()
    except Exception as e:
        print(f"Connection failed: {e}")
        return
    
    try:
        result = await runner.execute_tool(
            "filesystem",
            "read_file",
            {"path": "/nonexistent.txt"}
        )
    except RuntimeError as e:
        print(f"Tool execution failed: {e}")
    finally:
        await runner.disconnect_all()

Testing

# Install with dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Run with coverage
pytest --cov=tmcp_runner

Requirements

• Python 3.10+
• Dependencies (auto-installed):
  • mcp>=1.0.0 - Official Anthropic MCP SDK
  • httpx>=0.27.0 - HTTP client
  • httpx-sse>=0.4.0 - SSE support
  • pydantic>=2.0.0 - Data validation

Use Cases

Perfect for:

• 🤖 AI Agents - Give your AI agents access to tools and data sources
• 🌐 Web Applications - Add MCP capabilities to FastAPI, Flask, Django apps
• ⚙️ Background Services - Process MCP tasks asynchronously
• 🔄 Data Pipelines - Integrate MCP servers into ETL workflows
• 🧪 Testing Frameworks - Test MCP server implementations
• 📱 Desktop Applications - Build desktop apps with MCP integration

Security

• ✅ Process isolation for stdio servers
• ✅ Environment variable control
• ✅ No shell injection vulnerabilities
• ✅ Input validation
• ✅ Configurable permissions per server

Best Practices:

• Use restricted directories for filesystem access
• Use read-only tokens when possible
• Store secrets in environment variables
• Validate tool inputs before execution
• Review MCP server code before use

Documentation

• README.md (this file) - Library API and integration guide
• ARCHITECTURE.md - Technical architecture details
• example_usage.py - Comprehensive code examples
• demo.py - Interactive demonstration

Contributing

Contributions welcome! Please:

1. Fork the repository
2. Create a feature branch
3. Add tests for new features
4. Submit a pull request

License

MIT License - see LICENSE file for details

Resources

• MCP Documentation
• MCP Python SDK
• MCP Server Registry

Support

• 🐛 Issues: GitHub Issues
• 💬 Discussions: GitHub Discussions
• 📧 Email: [e-mail]

---

Made with ❤️ by Quantumics AI