Programmatic Usage

The server can be used programmatically in your Python applications.

Connection Providers

The server supports pluggable connection providers for different deployment scenarios:

from unblu_mcp import create_server, DefaultConnectionProvider

# Default provider (uses environment variables)
server = create_server()

# Custom provider with explicit credentials
provider = DefaultConnectionProvider(
    base_url="https://my-instance.unblu.cloud/app/rest/v4",
    api_key="my-api-key",
)
server = create_server(provider=provider)

Custom Connection Providers

Implement the ConnectionProvider interface for custom connectivity:

from unblu_mcp import ConnectionProvider, ConnectionConfig

class MyProvider(ConnectionProvider):
    async def setup(self) -> None:
        """Initialize connection (e.g., start tunnel)."""
        pass

    async def teardown(self) -> None:
        """Clean up resources."""
        pass

    async def ensure_connection(self) -> None:
        """Ensure connection is alive, restart if needed.

        Called before each API request. Override this if your
        provider needs to handle connection recovery.
        """
        pass

    def get_config(self) -> ConnectionConfig:
        """Return current connection configuration."""
        return ConnectionConfig(
            base_url="https://api.example.com",
            headers={"X-Custom-Header": "value"},
        )

    async def health_check(self) -> bool:
        """Check if connection is healthy."""
        return True

ConnectionConfig

The ConnectionConfig dataclass defines connection parameters:

from unblu_mcp import ConnectionConfig

config = ConnectionConfig(
    base_url="https://api.example.com/v4",
    headers={"X-Custom-Header": "value"},
    auth=("username", "password"),  # Optional basic auth tuple
    timeout=30.0,  # Request timeout in seconds
)

Running the Server

from unblu_mcp import create_server

server = create_server()

# Run with stdio transport (for MCP clients)
server.run()

# Or run with SSE transport for web clients
server.run(transport="sse", host="127.0.0.1", port=8000)

Custom OpenAPI Spec

You can provide a custom OpenAPI specification:

from unblu_mcp import create_server

server = create_server(spec_path="/path/to/custom-swagger.json")

With Policy Enforcement

from unblu_mcp import create_server

server = create_server(policy_file="/path/to/mcp_policies.json")

Feedback