mcpgateway-sdk 0.23.0

Python SDK for MCP Gateway — manage MCP servers, tools, skills, and sandboxes programmatically

mcpgateway-sdk

Python SDK for MCP Gateway — the enterprise platform for hosting, managing, and securing MCP (Model Context Protocol) servers.

Installation

pip install mcpgateway-sdk

Quick start

import asyncio
from mcpgateway_sdk import MCPGateway

async def main():
    async with MCPGateway(api_key="sk-...", url="http://localhost:8000") as gw:
        # Semantic search for tools across all servers
        tools = await gw.tools.search("get current weather")
        print(tools[0].tool_name, tools[0].score)

        # Execute a tool
        result = await gw.tools.execute(
            "get_weather", {"city": "NYC"}, server_name="weather"
        )
        print(result)

asyncio.run(main())

Features

• Servers — full CRUD, start/stop lifecycle, health checks, bulk actions, catalog import, tool overrides, credentials, OAuth config
• Tools — semantic search, execute by name, lookup by server/tool name
• Skills — create, upload, catalog import, attach/detach to servers
• Sessions — per-session tool scoping for multi-tenant isolation
• Sandboxes — isolated execution environments with file I/O and code exec
• Auth — API key management
• Cache — server-side cache control

All resources are accessed as attributes on the client: gw.servers, gw.tools, gw.skills, gw.sessions, gw.sandboxes, gw.auth, gw.cache.

Configuration

The client reads configuration from explicit parameters or environment variables:

Parameter	Environment variable	Default
api_key	MCPGATEWAY_TOKEN	—
url	MCPGATEWAY_URL	http://localhost:8000
timeout	—	30.0

import os
os.environ["MCPGATEWAY_TOKEN"] = "sk-..."
os.environ["MCPGATEWAY_URL"] = "https://gateway.example.com"

async with MCPGateway() as gw:  # picks up from env
    ...

Usage examples

All examples below assume gw is an active client from async with MCPGateway(...) as gw.

Manage servers

# List all running servers
servers = await gw.servers.list(status="running")

# Import a server from the built-in catalog
server = await gw.servers.import_from_catalog(
    registry="npm",
    registry_id="@anthropic/time-mcp",
    name="time-server",
)

# Start, stop, health check
await gw.servers.start(server.id)
health = await gw.servers.health(server.id)
await gw.servers.stop(server.id)

Search and execute tools

# Semantic search across all servers
results = await gw.tools.search("convert PDF to text", limit=3)
for r in results:
    print(f"{r.server_name}/{r.tool_name} — score: {r.score:.2f}")

# Execute a specific tool
output = await gw.tools.execute(
    "read_pdf",
    {"url": "https://example.com/report.pdf"},
    server_name="document-reader",
)

Work with skills

# Upload a skill from a local file
skill = await gw.skills.upload(file_path="./my_skill.md")

# Attach a skill to a server
await gw.skills.attach(skill_id=skill.id, server_id=server.id)

# Browse the skill catalog
entries = await gw.skills.catalog("builtin")

Sessions with scoped tools

# Allow specific tools + wildcard for entire servers
session = await gw.sessions.create(
    allowed_tool_names=["VVMCP__kb_finance", "HOTSPOT__*", "GMAIL__*"],
    denied_tool_names=["HOTSPOT__internal_debug"],
)

# Use the session ID for MCP connections with restricted tool access
print(session.id)

# Update only the deny list (allowed list unchanged)
session = await gw.sessions.update(
    session.id,
    denied_tool_names=["HOTSPOT__internal_debug", "HOTSPOT__admin_reset"],
)

# Refresh session activity
await gw.sessions.touch(session.id)

Sandbox execution

# Create a sandbox
sandbox = await gw.sandboxes.create(image="python:3.12")

# Execute code
result = await gw.sandboxes.exec(sandbox.id, command="python -c 'print(1+1)'")
print(result.stdout)

# Upload / download files
await gw.sandboxes.upload(sandbox.id, "data.csv", content=b"a,b\n1,2")
files = await gw.sandboxes.list_files(sandbox.id)

# Clean up
await gw.sandboxes.destroy(sandbox.id)

API key management

# Create a new API key
key = await gw.auth.create_api_key(name="ci-pipeline", expires_in_days=90)
print(key.key)  # Only shown once

# List and rotate keys
keys = await gw.auth.list_api_keys()
rotated = await gw.auth.rotate_api_key(keys[0].id)

SDK cache

# Store a value with TTL (default 300 seconds)
await gw.cache.set("last_run", {"status": "ok"}, ttl=600)

# Retrieve — returns None if key doesn't exist
value = await gw.cache.get("last_run")

Error handling

The SDK raises typed exceptions that map to HTTP status codes:

from mcpgateway_sdk import (
    GatewayError,    # Base — all API errors
    AuthError,       # 401 Unauthorized
    ForbiddenError,  # 403 Forbidden
    NotFoundError,   # 404 Not Found
    ConflictError,   # 409 Conflict
    ValidationError, # 422 Unprocessable Entity
    RateLimitError,  # 429 Too Many Requests
)

try:
    await gw.servers.get("nonexistent-id")
except NotFoundError:
    print("Server not found")
except GatewayError as e:
    print(f"API error [{e.status_code}]: {e.message}")

Version compatibility

The client automatically checks its version against the server on connect (via async with). If there is a major/minor mismatch, it emits a warning:

UserWarning: SDK version 0.16.0 does not match server version 0.17.0.
Update with: pip install --upgrade mcpgateway-sdk

Disable this check with MCPGateway(check_version=False).

MCP connection

Use the client to get connection details for direct MCP protocol access:

async with MCPGateway(api_key="sk-...") as gw:
    print(gw.mcp_url)       # http://localhost:8000/mcp/gateway
    print(gw.auth_headers)   # {"Authorization": "Bearer sk-..."}

Requirements

• Python 3.11+
• httpx >= 0.27
• Pydantic >= 2.0

License

MIT