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, typed remote registration, readiness, lifecycle, health checks, bulk actions, catalog import, tool sync (with bundle-update control), tool overrides, credentials, OAuth config
• Tools — semantic search, execute by name, lookup by server/tool name
• Skills — create, upload, catalog import, package download
• Generation — spec import without AI (servers.import_schema) and admin AI generation jobs from a URL, docs or text (generation), with fail-closed bundle_only
• Script tools — upload a server-owned script-tool package (servers.create_script_tool_package)
• 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, gw.oauth_apps, gw.oauth_templates, and gw.generation.

Configuration

The client reads configuration from explicit parameters or environment variables:

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)

# Re-sync tools from the source. By default, new/renamed tools are also
# propagated into any bundles that curate this server. Check how many bundles
# would be affected, then opt out of the bundle update if you prefer.
usage = await gw.servers.bundle_usage(server.id)  # {"bundle_count": 3}
await gw.servers.sync_tools(server.id)                       # update bundles (default)
await gw.servers.sync_tools(server.id, update_bundles=False) # leave bundles alone

# Uploading a script-tool package works the same way (default on):
await gw.servers.create_script_tool_package(server.id, "package.zip")
await gw.servers.create_script_tool_package(server.id, "package.zip", update_bundles=False)

Create a server from an API spec or docs

# Deterministic, no AI: a pasted OpenAPI/Swagger spec (JSON or YAML).
# auth_mode="none" accepts any security scheme; declare auth afterwards with
# gw.servers.authentication(server.id).set(...). bundle_only needs admin/owner.
server = await gw.servers.import_schema(
    "acme-crm", spec_text, display_name="Acme CRM", auth_mode="none", access_mode="bundle_only"
)

AI generation from a URL, docs or free text is an admin-only job: start it, poll until it is reviewing, inspect the analysis, then approve to deploy.

import asyncio

from mcpgateway_sdk import (
    AccessModeNotEnforcedError,
    GatewayCapabilityError,
    GenerationJobExistsError,
)
from mcpgateway_sdk.models import TERMINAL_GENERATION_STATUSES

try:
    job = await gw.generation.start(
        "url", "https://acme.dev/openapi.json", "acme-api", access_mode="bundle_only"
    )
    job_id = job.job_id
except GenerationJobExistsError as exc:  # e.g. retry after a timeout
    if exc.existing_job_id is None:
        raise  # the gateway didn't say which job: look it up with gw.generation.list()
    job_id = exc.existing_job_id  # resume it, but see expected_access_mode below
except GatewayCapabilityError:  # gateway too old or /health unreachable: nothing was created
    raise
except AccessModeNotEnforcedError as exc:  # created without bundle_only; cleanup attempted
    if not exc.cleaned_up:
        print(f"remove job {exc.job_id} by hand: {exc.cleanup}")
    raise

while (status := await gw.generation.get(job_id)).status != "reviewing":
    if status.status in TERMINAL_GENERATION_STATUSES:  # deployed, failed or cancelled
        raise RuntimeError(f"job {job_id} ended {status.status}: {status.error_message}")
    await asyncio.sleep(5)
analysis = await gw.generation.analysis(job_id)  # planned tools, base URL, auth
# expected_access_mode: a resumed job may have been created PUBLIC (an older pod
# mid-upgrade, or someone else's job); approve() then refuses instead of deploying it.
# credential_mode="none": deploy without a credential shape, then declare the real one
# with gw.servers.authentication(server_id).set(...) (analysis["auth"] pre-fills it).
await gw.generation.approve(job_id, credential_mode="none", expected_access_mode="bundle_only")

access_mode="bundle_only" fails closed. Before sending anything, start() checks that GET /health lists the bundle_only_generation capability, so the gateway's /health must be reachable from the client; if it isn't, start() raises GatewayCapabilityError with reason="unreachable". If the gateway is too old or the answer can't be parsed, it raises the same error (missing / unparseable) and no job is created; a confirmed capability is remembered per client, a failure is re-checked next time. The advisory version warning from check_version() is not a substitute. If a job is nevertheless created without the requested access mode (a rolling upgrade can answer /health from a new pod and the POST from an old one), start() cancels and deletes it and raises AccessModeNotEnforcedError; exc.cleaned_up is True once the job can no longer deploy. A create response the SDK can't read raises GenerationOutcomeUnknownError (a job may exist; check gw.generation.list()). Always approve a job you didn't just create with expected_access_mode. access_mode="public" (the default) needs no check.

Register an authenticated remote MCP server

from mcpgateway_sdk.models import RemoteAuthentication, RemoteServerRegistration, ServerReadiness

# Probe first: only OAuth and no-auth can be safely preselected by the gateway.
probe = await gw.oauth_apps.probe_auth("https://mcp.example.com/mcp")
print(probe.outcome)

server = await gw.servers.register_remote(
    RemoteServerRegistration(
        name="example-remote",
        display_name="Example Remote",
        url="https://mcp.example.com/mcp",
        authentication=RemoteAuthentication(
            type="api_key",
            config={"location": "header", "name": "X-API-Key"},
            secrets={"api_key": "set-once-and-never-read-back"},
        ),
    )
)
if server.readiness is ServerReadiness.AWAITING_CREDENTIALS:
    print("A member still needs to connect their account")
elif server.readiness is ServerReadiness.FAILED:
    print(server.last_error)

The typed remote path supports only none, api_key, bearer, basic, and oauth2. It rejects HMAC and query-parameter API keys locally. servers.create(**fields) remains available for existing generic server workflows.

For OAuth, list a template and create the app before registering the server; the app response's id is the config["app_id"] value for RemoteAuthentication(type="oauth2", ...):

import os

from mcpgateway_sdk.models import OAuthAppCreate

template = (await gw.oauth_templates.list())[0]
app = await gw.oauth_apps.create(
    OAuthAppCreate(
        alias="example-oauth",
        name="Example OAuth",
        template_id=template.id,
        client_id="provider-issued-client-id",
        client_secret=os.environ["OAUTH_CLIENT_SECRET"],
    )
)
print(app.id)  # Pass this UUID as the remote OAuth2 config.app_id.

For an API integration (a bundle_only virtual server), create a template-less custom app instead, register its callback_url at the provider, declare OAuth on the server, and run the consent flow once. Use the SAME API key (an admin's) for these calls and for every agent's tools/execute (with the agent's bundle_id): the connection belongs to that key's owner, and a connection made by another user, e.g. in the portal, is not used. Keep the connection id to reconnect in place (see docs/guides/api-integration-credentials.md):

from mcpgateway_sdk.models import OAuthAppCustomCreate, OAuthAppUpdate

app = await gw.oauth_apps.create_custom(
    OAuthAppCustomCreate(
        alias="acme-crm",
        name="Acme CRM",
        client_id="provider-client-id",
        client_secret=os.environ["OAUTH_CLIENT_SECRET"],  # omit for a public client
        token_endpoint_auth_method="client_secret_post",  # or "client_secret_basic" / "none"
        auth_url="https://auth.acme.example/oauth/authorize",
        token_url="https://auth.acme.example/oauth/token",
    )
)
print(app.callback_url)  # the redirect URI to register at the provider

await gw.servers.authentication(server_id).set(
    "oauth2", {"app_id": str(app.id), "scopes": ["crm.read"]}, []
)
auth = await gw.connections.initiate_oauth(
    server_id, app.id, redirect_uri="https://vivi.example/integrations/done"
)
print(auth.authorization_url)  # send the end user here

# Later: reconnect the same connection instead of creating a second one.
(conn,) = await gw.connections.list(server_id=server_id, include_invalid=True)
auth = await gw.connections.initiate_oauth(server_id, app.id, connection_id=conn.id)

# Rotating the secret keeps connections; changing client_id or an endpoint deletes them.
app = await gw.oauth_apps.update(app.id, OAuthAppUpdate(client_secret="new-secret"))
same = await gw.oauth_apps.get(app.id)
# gw.servers.delete(server_id) also deletes a custom app no other server uses; delete
# one explicitly only to drop it while the server stays.
await gw.oauth_apps.delete(app.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")

# 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_file(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
    GenerationJobExistsError,  # 409: a live generation job exists (.existing_job_id)
    GenerationJobBusyError,    # 409: a deploy holds the job; retry shortly (.retryable)
)

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