MCP Servers - Python SDK

Docs

Manage Model Context Protocol servers with the Python SDK

MCP Server Management

The Model Context Protocol (MCP) allows you to extend AI capabilities with custom tools and data sources. The SDK provides full management of MCP servers.

Create MCP Server

from lumnisai import Client
 
client = Client(api_key="your-api-key")
 
# Create stdio-based MCP server
server = client.create_mcp_server(
    name="github-tools",
    description="GitHub API tools",
    transport="stdio",
    scope="tenant",
    command="npx",
    args=["-y", "@modelcontextprotocol/server-github"],
    env={
        "GITHUB_TOKEN": "your-github-token"
    }
)
 
print(f"Server created: {server.server_id}")
print(f"Name: {server.name}")
print(f"Transport: {server.transport}")

Create HTTP-based MCP Server

# Create HTTP/SSE-based MCP server
server = client.create_mcp_server(
    name="custom-api-tools",
    description="Custom API integration",
    transport="streamable_http",
    scope="tenant",
    url="https://mcp-server.example.com/api",
    headers={
        "Authorization": "Bearer token",
        "X-API-Key": "api-key"
    }
)

Create User-Scoped MCP Server

# Create MCP server for specific user
user_server = client.create_mcp_server(
    name="personal-supabase",
    description="User's personal Supabase instance",
    transport="stdio",
    scope="user",
    user_identifier="user@example.com",
    command="npx",
    args=["-y", "@supabase/mcp-server-supabase@latest", "--project-ref=abc123"],
    env={
        "SUPABASE_ACCESS_TOKEN": "user-token"
    }
)

List MCP Servers

# List all servers
servers = client.list_mcp_servers(limit=100)
 
print(f"Total servers: {len(servers.servers)}")
 
for server in servers.servers:
    print(f"\n{server.name}")
    print(f"  ID: {server.server_id}")
    print(f"  Transport: {server.transport}")
    print(f"  Scope: {server.scope}")
    print(f"  Active: {server.is_active}")

List with Filters

# Filter by scope
tenant_servers = client.list_mcp_servers(
    scope="tenant",
    is_active=True,
    skip=0,
    limit=50
)
 
# Filter by user
user_servers = client.list_mcp_servers(
    scope="user",
    user_identifier="user@example.com",
    is_active=True
)
 
# Get all servers (tenant + user)
all_servers = client.list_mcp_servers(scope="all")

Get Server Details

# Get specific server
server = client.get_mcp_server(server_id="server-id")
 
print(f"Name: {server.name}")
print(f"Description: {server.description}")
print(f"Transport: {server.transport}")
print(f"Active: {server.is_active}")
 
if server.transport == "stdio":
    print(f"Command: {server.command}")
    print(f"Args: {server.args}")
elif server.transport in ["streamable_http", "sse"]:
    print(f"URL: {server.url}")

Update MCP Server

# Update server configuration
updated = client.update_mcp_server(
    server_id="server-id",
    description="Updated description",
    is_active=True,
    env={
        "NEW_VAR": "new-value"
    }
)
 
print(f"Updated: {updated.name}")

Delete MCP Server

# Delete server
client.delete_mcp_server(server_id="server-id")
print("Server deleted")

Test MCP Server Connection

# Test server connection and list tools
test_result = client.test_mcp_server(server_id="server-id")
 
print(f"Connection: {'Success' if test_result.success else 'Failed'}")
print(f"Tools found: {test_result.tool_count}")
 
if test_result.error:
    print(f"Error: {test_result.error}")

List Server Tools

# Get tools provided by MCP server
tools = client.list_mcp_server_tools(server_id="server-id")
 
print(f"Server provides {len(tools.tools)} tools:")
 
for tool in tools.tools:
    print(f"\n{tool.name}")
    print(f"  Description: {tool.description}")
    if tool.input_schema:
        print(f"  Parameters: {tool.input_schema.get('properties', {}).keys()}")

Complete MCP Server Setup

from lumnisai import Client
import os
 
client = Client(api_key="your-api-key")
 
# Step 1: Create MCP server
print("Creating MCP server...")
server = client.create_mcp_server(
    name="supabase-mcp",
    description="Supabase database integration",
    transport="stdio",
    scope="tenant",
    command="npx",
    args=[
        "-y",
        "@supabase/mcp-server-supabase@latest",
        "--project-ref=your-project-ref"
    ],
    env={
        "SUPABASE_ACCESS_TOKEN": os.getenv("SUPABASE_ACCESS_TOKEN")
    }
)
 
print(f"✓ Server created: {server.server_id}")
 
# Step 2: Test connection
print("\nTesting connection...")
test = client.test_mcp_server(server.server_id)
 
if test.success:
    print(f"✓ Connection successful!")
    print(f"✓ Found {test.tool_count} tools")
else:
    print(f"✗ Connection failed: {test.error}")
    # Optionally delete and recreate
    client.delete_mcp_server(server.server_id)
    exit(1)
 
# Step 3: List available tools
print("\nAvailable tools:")
tools = client.list_mcp_server_tools(server.server_id)
 
for tool in tools.tools:
    print(f"  • {tool.name}: {tool.description}")
 
# Step 4: Use in AI response
print("\nUsing MCP server in AI task...")
response = client.invoke(
    "Query the database for user statistics",
    user_id="user@example.com"
)
 
print(f"\nResult: {response.output_text}")
 
# Step 5: List all servers
print("\nAll configured MCP servers:")
all_servers = client.list_mcp_servers()
 
for s in all_servers.servers:
    status = "active" if s.is_active else "inactive"
    print(f"  • {s.name} ({status})")

Using MCP Servers with AI

Automatic Tool Discovery

from lumnisai import Client, AgentConfig
 
client = Client(api_key="your-api-key")
 
# Create response - MCP server tools automatically available
response = client.invoke(
    """
    Create a new GitHub repository called "my-project"
    and create an initial README.md file
    """,
    user_id="user@example.com",
    agent_config=AgentConfig(
        coordinator_model_name="openai:gpt-4.1",
        use_cognitive_tools=True
    )
)
 
print(response.output_text)

With Streaming

from lumnisai import Client, display_progress
 
client = Client(api_key="your-api-key")
 
for update in client.invoke(
    "Search my Supabase database for recent orders",
    stream=True,
    user_id="user@example.com"
):
    display_progress(update)
    
    # See which MCP tools are being used
    if update.tool_calls:
        for tool in update.tool_calls:
            if tool.get('name', '').startswith('supabase_'):
                print(f"  → Using Supabase tool: {tool.get('name')}")

Common MCP Server Examples

GitHub MCP Server

github_server = client.create_mcp_server(
    name="github-integration",
    description="GitHub repository management",
    transport="stdio",
    scope="tenant",
    command="npx",
    args=["-y", "@modelcontextprotocol/server-github"],
    env={
        "GITHUB_TOKEN": os.getenv("GITHUB_TOKEN")
    }
)

Supabase MCP Server

supabase_server = client.create_mcp_server(
    name="supabase-db",
    description="Supabase database access",
    transport="stdio",
    scope="user",
    user_identifier="user@example.com",
    command="npx",
    args=[
        "-y",
        "@supabase/mcp-server-supabase@latest",
        "--project-ref=your-project-ref"
    ],
    env={
        "SUPABASE_ACCESS_TOKEN": os.getenv("SUPABASE_ACCESS_TOKEN")
    }
)

Custom HTTP MCP Server

custom_server = client.create_mcp_server(
    name="internal-api",
    description="Internal company API",
    transport="streamable_http",
    scope="tenant",
    url="https://mcp.company.com/api",
    headers={
        "Authorization": f"Bearer {os.getenv('INTERNAL_API_TOKEN')}",
        "X-Tenant-ID": "tenant-123"
    }
)

Async MCP Server Management

from lumnisai import AsyncClient
import asyncio
 
async def setup_mcp_servers():
    client = AsyncClient(api_key="your-api-key")
    
    async with client:
        # Create multiple servers concurrently
        servers = await asyncio.gather(
            client.create_mcp_server(
                name="github",
                transport="stdio",
                scope="tenant",
                command="npx",
                args=["-y", "@modelcontextprotocol/server-github"],
                env={"GITHUB_TOKEN": os.getenv("GITHUB_TOKEN")}
            ),
            client.create_mcp_server(
                name="slack",
                transport="stdio",
                scope="tenant",
                command="npx",
                args=["-y", "@modelcontextprotocol/server-slack"],
                env={"SLACK_TOKEN": os.getenv("SLACK_TOKEN")}
            )
        )
        
        # Test all servers concurrently
        test_results = await asyncio.gather(*[
            client.test_mcp_server(server.server_id)
            for server in servers
        ])
        
        for server, result in zip(servers, test_results):
            status = "✓" if result.success else "✗"
            print(f"{status} {server.name}: {result.tool_count} tools")
 
asyncio.run(setup_mcp_servers())

Best Practices

Test Before Using

def create_and_test_mcp_server(client, config):
    """Create MCP server and verify it works."""
    # Create server
    server = client.create_mcp_server(**config)
    
    # Test connection
    test = client.test_mcp_server(server.server_id)
    
    if not test.success:
        # Clean up failed server
        client.delete_mcp_server(server.server_id)
        raise Exception(f"MCP server test failed: {test.error}")
    
    print(f"Server {server.name} ready with {test.tool_count} tools")
    return server

Handle Environment Variables Securely

import os
 
# Load from environment
server = client.create_mcp_server(
    name="github",
    transport="stdio",
    scope="tenant",
    command="npx",
    args=["-y", "@modelcontextprotocol/server-github"],
    env={
        "GITHUB_TOKEN": os.getenv("GITHUB_TOKEN"),  # Never hardcode tokens
    }
)

Disable Instead of Delete

# Temporarily disable instead of deleting
client.update_mcp_server(
    server_id="server-id",
    is_active=False
)
 
# Re-enable later
client.update_mcp_server(
    server_id="server-id",
    is_active=True
)

Monitor Server Health

def check_mcp_server_health(client, server_id):
    """Verify MCP server is healthy."""
    try:
        test = client.test_mcp_server(server_id)
        
        if test.success and test.tool_count > 0:
            return True
        else:
            print(f"Server unhealthy: {test.error}")
            return False
    except Exception as e:
        print(f"Health check failed: {e}")
        return False
 
# Usage
if not check_mcp_server_health(client, "server-id"):
    # Restart or recreate server
    print("Server needs attention")

Integrations Model Preferences