coreason-mcp 0.4.0

Standardizes external data access using the Model Context Protocol.

coreason-mcp

The Complete Gateway for the Model Context Protocol (MCP).

coreason-mcp serves as the Universal Interface Layer for the CoReason ecosystem. It replaces ad-hoc API wrappers with a standardized Model Context Protocol (MCP) Gateway, acting as an MCP Host that connects to both remote services (via SSE) and local sidecars (via Stdio).

Installation

pip install coreason-mcp

Features

• Hybrid Transport: Support for both SSE (remote microservices) and Stdio (local sidecars) transports.
• Tool Aggregation: Merges tools from multiple servers into a unified namespace with automatic conflict resolution.
• Secure Credential Injection: Manages API keys and headers securely in the transport layer, keeping them out of the agent's context.
• Async/High Concurrency: Built on httpx and asyncio to ensure non-blocking operations across the mesh.
• Robust Error Handling: Maps connection and protocol errors to standardized exceptions.

For detailed requirements and architecture, see Product Requirements.

Usage

Here is how to initialize and use the library to connect to a local filesystem tool and a remote database.

import asyncio
from coreason_mcp.config import McpServerConfig, TransportType
from coreason_mcp.manager import McpManager

# 1. Configure your servers (Mesh Topology)
# A local sidecar (Stdio) - e.g., for secure filesystem access
local_fs = McpServerConfig(
    name="local_fs",
    transport=TransportType.STDIO,
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem", "/home/user/docs"]
)

# A remote microservice (SSE) - e.g., for clinical data
clinical_db = McpServerConfig(
    name="clinical_db",
    transport=TransportType.SSE,
    url="http://db-service.internal:8080/sse",
    auth_headers={"Authorization": "Bearer s3cr3t-t0k3n"}
)

# 2. Initialize the manager
manager = McpManager([local_fs, clinical_db])

async def main():
    # 3. Connect to all servers
    await manager.connect_all()

    try:
        # 4. List aggregated tools (converted to OpenAI format)
        tools = await manager.get_openai_tools()
        print(f"Available tools: {[t['function']['name'] for t in tools]}")

        # 5. Call a tool
        # The manager handles routing and error mapping
        result = await manager.call_tool(
            tool_name="clinical_db_get_patient",
            arguments={"patient_id": "PT-12345"}
        )
        print("Result:", result.content[0].text)

    finally:
        # 6. Cleanup connections
        await manager.close()

if __name__ == "__main__":
    asyncio.run(main())