xylent-mcp-python 0.1.2

Production-ready MCP (Model Context Protocol) client and server implementation for Xylent services

xylent-mcp-python

Production-ready MCP (Model Context Protocol) client and server implementation for Python, featuring robust client wrapper and FastAPI server integration.

Features

Client Features

• 🚀 Lazy connection establishment - connects only when needed
• 🔄 Automatic retry with exponential backoff for transient failures
• 🔐 Pluggable authentication strategies (Basic, Token, Custom)
• 📦 Singleton registry with per-URL connection pooling
• 🛡️ Type-safe with full typing support
• ⚡ Async/await native implementation with concurrent optimization
• 🎯 Comprehensive error taxonomy for better error handling

Server Features (v0.1.0+)

• 🌐 FastAPI integration via attach_mcp() for serving MCP tools
• 🔒 Production-ready BasicAuth for MCP server authentication
• 🤝 Seamless client-server authentication alignment
• 🛡️ CORS support and exception handling
• 📊 Path allowlisting for health checks and metrics

Requirements

• Python 3.11+ (Python 3.13 recommended)

Installation

Basic Installation (Client Only)

pip install xylent-mcp-python

With Server Support

pip install xylent-mcp-python[server]

This installs FastAPI, Starlette, and Uvicorn for serving MCP endpoints.

Using Poetry

# Client only
poetry add xylent-mcp-python

# With server support
poetry add xylent-mcp-python[server]

Quick Start

Client Usage

import asyncio
from xylent.mcp_python import MCPClientRegistry
from xylent.mcp_python.auth import BasicHTTPAuth

async def main():
    # Create client with authentication
    auth = BasicHTTPAuth("username", "password")
    client = await MCPClientRegistry.get(
        "http://localhost:8000/mcp",
        auth=auth
    )

    # List available tools
    tools = await client.list_tools()
    print(f"Available tools: {tools}")

    # Call a tool
    result = await client.call_tool(
        "calculator",
        arguments={"operation": "add", "a": 5, "b": 3}
    )
    print(f"Result: {result}")

    # Cleanup
    await MCPClientRegistry.aclose_all()

asyncio.run(main())

Server Usage (FastAPI)

from fastapi import FastAPI
from xylent.mcp_python.serverkit import attach_mcp
from pydantic import BaseModel, Field

app = FastAPI()

# Define tool input schema
class CalculatorInput(BaseModel):
    operation: str = Field(description="Operation to perform")
    a: float = Field(description="First number")
    b: float = Field(description="Second number")

# Define tool handler
async def calculator_tool(operation: str, a: float, b: float):
    operations = {
        "add": a + b,
        "subtract": a - b,
        "multiply": a * b,
        "divide": a / b,
    }
    return {
        "content": [{"type": "text", "text": str(operations[operation])}]
    }

# Attach MCP server
attach_mcp(
    app,
    name="calculator-service",
    version="1.0.0",
    tools=[
        {
            "name": "calculator",
            "description": "Perform arithmetic operations",
            "inputSchema": CalculatorInput,
            "handler": calculator_tool,
        }
    ],
    # Auth defaults to BasicAuth from BASIC_USERNAME/BASIC_PASSWORD env vars
    # Or explicitly: auth=BasicAuth(username="user", password="pass")
)

# Run with: uvicorn main:app --host 0.0.0.0 --port 8000

Authentication

Client-Side Authentication

Basic HTTP Authentication

from xylent.mcp_python.auth import BasicHTTPAuth

auth = BasicHTTPAuth("username", "password")
client = await MCPClientRegistry.get(server_url, auth=auth)

No Authentication

from xylent.mcp_python.auth import NoAuth

client = await MCPClientRegistry.get(server_url, auth=NoAuth())

Custom Authentication

from xylent.mcp_python.auth import AuthStrategy

class CustomAuth(AuthStrategy):
    def get_auth_headers(self) -> dict[str, str] | None:
        return {"Authorization": f"Bearer {self.token}"}

client = await MCPClientRegistry.get(server_url, auth=CustomAuth())

Server-Side Authentication

The attach_mcp function provides smart authentication defaults:

Default: Environment Variables

# Reads BASIC_USERNAME and BASIC_PASSWORD from environment
# Raises ValueError if not set (fail-safe)
attach_mcp(app, name="my-service", tools=[...])

Explicit BasicAuth

from xylent.mcp_python.serverkit.auth import BasicAuth

attach_mcp(
    app,
    name="my-service",
    tools=[...],
    auth=BasicAuth(username="custom-user", password="custom-pass")
)

Explicit NoAuth (Development Only)

from xylent.mcp_python.serverkit.auth import NoAuth

attach_mcp(
    app,
    name="my-service",
    tools=[...],
    auth=NoAuth()  # Explicitly disable auth
)

Server Integration

FastAPI Integration

from fastapi import FastAPI
from xylent.mcp_python.serverkit import attach_mcp

app = FastAPI()

# Initialize your services
async def app_context():
    """Initialize services once on startup"""
    # Connect to database
    # Initialize API clients
    # Setup caches
    pass

attach_mcp(
    app,
    name="my-service",
    version="1.0.0",
    tools=[...],
    app_context=app_context,
    path="/mcp",
    cors=True,
    allow_paths=["/health", "/metrics"],
)

# Regular FastAPI endpoints work normally
@app.get("/health")
async def health():
    return {"status": "healthy"}

Tool Definition

Tools are defined with Pydantic schemas for automatic validation:

from pydantic import BaseModel, Field

class QueryInput(BaseModel):
    query: str = Field(description="SQL query to execute")
    limit: int = Field(default=100, description="Result limit")

async def query_tool(query: str, limit: int = 100):
    # Tool implementation
    results = await db.execute(query, limit=limit)

    return {
        "content": [
            {"type": "text", "text": str(results)}
        ]
    }

# Register tool
tools = [
    {
        "name": "query_database",
        "description": "Execute a database query",
        "inputSchema": QueryInput,
        "handler": query_tool,
    }
]

Resource Definition

Expose data endpoints with resources:

async def config_resource(uri: str):
    """Return application configuration"""
    return {
        "contents": [
            {
                "uri": uri,
                "text": '{"setting": "value"}',
                "mimeType": "application/json",
            }
        ]
    }

resources = [
    {
        "name": "app_config",
        "uri": "config://app/settings",
        "description": "Application configuration",
        "mimeType": "application/json",
        "handler": config_resource,
    }
]

attach_mcp(app, name="my-service", resources=resources)

Error Handling

from xylent.mcp_python import MCPError, MCPConnectionError, MCPTimeout

try:
    result = await client.call_tool("my-tool", arguments={"arg": "value"})
except MCPConnectionError as e:
    print(f"Connection failed: {e}")
except MCPTimeout as e:
    print(f"Operation timed out: {e}")
except MCPError as e:
    print(f"MCP error: {e}")

# Server-side error handling
def handle_error(error: Exception):
    logger.error(f"[MCP] Error: {error}")

attach_mcp(
    app,
    name="my-service",
    tools=[...],
    on_exception=handle_error
)

App Context Pattern

Initialize services once on startup:

from langfuse import Langfuse
from sqlalchemy.ext.asyncio import create_async_engine

# Global context
app_context = None

async def initialize_app_context():
    """Initialize services once on startup"""
    global app_context

    # Initialize database
    engine = create_async_engine("postgresql+asyncpg://...")

    # Initialize Langfuse
    langfuse = Langfuse()

    # Initialize other services
    app_context = {
        "db": engine,
        "langfuse": langfuse,
    }

    print("[Init] Services initialized")

# Use in tool handlers
async def query_tool(query: str):
    engine = app_context["db"]
    async with engine.connect() as conn:
        result = await conn.execute(query)
        return {"content": [{"type": "text", "text": str(result)}]}

attach_mcp(
    app,
    name="my-service",
    tools=[...],
    app_context=initialize_app_context
)

Registry Pattern

The MCPClientRegistry implements a singleton pattern to ensure one client per URL:

from xylent.mcp_python import MCPClientRegistry

# Get or create client (idempotent)
client1 = await MCPClientRegistry.get(url, auth=auth)
client2 = await MCPClientRegistry.get(url, auth=auth)  # Returns same instance

# Remove specific client
await MCPClientRegistry.remove_client(url)

# Close all clients (call during shutdown)
await MCPClientRegistry.aclose_all()

Configuration

Client Configuration

from xylent.mcp_python import MCPClientConfig

config = MCPClientConfig(
    server_url="http://localhost:8000/mcp",
    auth_type="basic",
    basic_username="user",
    basic_password="pass",
    connect_timeout_ms=10000,
    stream_read_timeout_ms=60000,
    max_attempts=3,
    retry_min_wait_s=0.5,
    retry_max_wait_s=5.0,
)

Server Configuration

from xylent.mcp_python.serverkit import attach_mcp
from xylent.mcp_python.serverkit.auth import BasicAuth

attach_mcp(
    app,
    name="my-service",           # Required: Service name
    version="1.0.0",             # Optional: Service version
    tools=[...],                 # Optional: List of tools
    resources=[...],             # Optional: List of resources
    app_context=init_function,   # Optional: Async init function
    auth=BasicAuth(...),         # Optional: Auth backend (defaults to env)
    path="/mcp",                 # Optional: Mount path (default: /mcp)
    cors=True,                   # Optional: Enable CORS (default: False)
    cors_origins=["*"],          # Optional: Specific CORS origins
    allow_paths=["/health"],     # Optional: Paths to exclude from auth
    on_exception=error_handler,  # Optional: Error callback
)

Examples

See the examples directory for complete working examples:

• attach_mcp_example.py: Full FastAPI integration with auth
• simple_initialization.py: Client initialization patterns
• strict_auth_example.py: Authentication examples

Development

# Install with Poetry
poetry install

# Install with server extras
poetry install -E server

# Run tests
poetry run pytest

# Run tests with coverage
poetry run pytest --cov=src/xylent

# Lint
poetry run ruff check src tests

# Format
poetry run ruff format src tests

# Type check
poetry run mypy src

License

Proprietary - Xylent Internal Use Only

Contributing

This is a proprietary package for internal use. External contributions are not accepted.

Support

For internal support, please contact:

• Santiago Blankleider [e-mail]

Changelog

See CHANGELOG.md for version history.

Links

• MCP Protocol Specification
• FastAPI Documentation
• Pydantic Documentation