official_registry_client

Class info

Classes

Name	Children	Inherits
GetServerCacheEntry llmling_agent.mcp_server.registries.official_registry_client Cache entry for get_server results.		object
ListServersCacheEntry llmling_agent.mcp_server.registries.official_registry_client Cache entry for list_servers results.		object
MCPRegistryClient llmling_agent.mcp_server.registries.official_registry_client Client for interacting with the MCP registry API.		object
MCPRegistryError llmling_agent.mcp_server.registries.official_registry_client Exception raised for MCP registry operations.		Exception
RegistryListResponse llmling_agent.mcp_server.registries.official_registry_client Response from the registry list servers endpoint.		Schema
RegistryPackage llmling_agent.mcp_server.registries.official_registry_client Package information for installing an MCP server.		Schema
RegistryRemote llmling_agent.mcp_server.registries.official_registry_client Remote endpoint configuration.		Schema
RegistryRepository llmling_agent.mcp_server.registries.official_registry_client Repository information for a registry server.		Schema
RegistryServer llmling_agent.mcp_server.registries.official_registry_client MCP server entry from the registry.		Schema
RegistryServerWrapper llmling_agent.mcp_server.registries.official_registry_client Wrapper for server data from the official registry API.		Schema
RegistryTransport llmling_agent.mcp_server.registries.official_registry_client Transport configuration for a package.		Schema
UnsupportedTransportError llmling_agent.mcp_server.registries.official_registry_client Raised when no supported transport is available.		Exception

🛈 DocStrings

MCP Registry client service for discovering and managing MCP servers.

This module provides functionality to interact with the Model Context Protocol registry API for server discovery and configuration.

GetServerCacheEntry dataclass

Cache entry for get_server results.

Source code in src/llmling_agent/mcp_server/registries/official_registry_client.py

@dataclass(slots=True)
class GetServerCacheEntry:
    """Cache entry for get_server results."""

    server: RegistryServer
    timestamp: float

ListServersCacheEntry dataclass

Cache entry for list_servers results.

Source code in src/llmling_agent/mcp_server/registries/official_registry_client.py

@dataclass(slots=True)
class ListServersCacheEntry:
    """Cache entry for list_servers results."""

    servers: list[RegistryServer]
    timestamp: float

MCPRegistryClient

Client for interacting with the MCP registry API.

Source code in src/llmling_agent/mcp_server/registries/official_registry_client.py

class MCPRegistryClient:
    """Client for interacting with the MCP registry API."""

    def __init__(self, base_url: str = "https://registry.modelcontextprotocol.io") -> None:
        self.base_url = base_url.rstrip("/")
        self.client = httpx.AsyncClient(timeout=30.0)
        self._cache_lists: dict[str, ListServersCacheEntry] = {}
        self._cache_servers: dict[str, GetServerCacheEntry] = {}

    async def list_servers(
        self, search: str | None = None, status: str = "active"
    ) -> list[RegistryServer]:
        """List servers from registry with optional filtering."""
        cache_key = f"list_servers:{search}:{status}"

        # Check cache first
        if cache_key in self._cache_lists:
            cached_entry = self._cache_lists[cache_key]
            if time.time() - cached_entry.timestamp < CACHE_TTL:
                log.debug("Using cached server list")
                return cached_entry.servers

        try:
            log.info("Fetching server list from registry")
            response = await self.client.get(f"{self.base_url}/v0/servers")
            response.raise_for_status()
            data = response.json()
        except httpx.HTTPError as e:
            msg = f"Failed to list servers: {e}"
            raise MCPRegistryError(msg) from e
        else:
            response_data = RegistryListResponse(**data)
            wrappers = response_data.servers

            if status:  # Filter by status from metadata
                wrappers = [
                    wrapper
                    for wrapper in wrappers
                    if wrapper.meta.get("io.modelcontextprotocol.registry/official", {}).get(
                        "status"
                    )
                    == status
                ]

            servers = [wrapper.server for wrapper in wrappers]
            if search:  # Filter by search term
                search_lower = search.lower()
                servers = [
                    s
                    for s in servers
                    if search_lower in s.name.lower() or search_lower in s.description.lower()
                ]

            # Cache the result
            self._cache_lists[cache_key] = ListServersCacheEntry(
                servers=servers, timestamp=time.time()
            )
            log.info("Successfully fetched %d servers", len(servers))
            return servers

    async def get_server(self, server_id: str) -> RegistryServer:
        """Get full server details including packages."""
        cache_key = f"server:{server_id}"

        # Check cache first
        if cache_key in self._cache_servers:
            cached_entry = self._cache_servers[cache_key]
            if time.time() - cached_entry.timestamp < CACHE_TTL:
                log.debug("Using cached server details for %s", server_id)
                return cached_entry.server

        try:
            log.info("Fetching server details for %s", server_id)

            # Get all wrappers to access metadata
            response = await self.client.get(f"{self.base_url}/v0/servers")
            response.raise_for_status()
            data = response.json()
            response_data = RegistryListResponse(**data)

            # Find server by name
            target_wrapper = None
            for wrapper in response_data.servers:
                if wrapper.server.name == server_id:
                    target_wrapper = wrapper
                    break

            if not target_wrapper:
                msg = f"Server {server_id!r} not found in registry"
                raise MCPRegistryError(msg)

            # Get the UUID from metadata
            server_uuid = target_wrapper.meta.get(
                "io.modelcontextprotocol.registry/official", {}
            ).get("id")

            if not server_uuid:
                msg = f"No UUID found for server {server_id!r}"
                raise MCPRegistryError(msg)

            # Now fetch the full server details using UUID
            response = await self.client.get(f"{self.base_url}/v0/servers/{server_uuid}")
            response.raise_for_status()

            server_data = response.json()
            server = RegistryServer(**server_data)

            # Cache the result
            self._cache_servers[cache_key] = GetServerCacheEntry(
                server=server, timestamp=time.time()
            )
            log.info("Successfully fetched server details for %s", server_id)

        except httpx.HTTPStatusError as e:
            if e.response.status_code == HTTP_NOT_FOUND:
                msg = f"Server {server_id!r} not found in registry"
                raise MCPRegistryError(msg) from e
            msg = f"Failed to get server details: {e}"
            raise MCPRegistryError(msg) from e
        except (httpx.HTTPError, ValueError, KeyError) as e:
            msg = f"Failed to get server details: {e}"
            raise MCPRegistryError(msg) from e
        else:
            return server

    def clear_cache(self) -> None:
        """Clear the metadata cache."""
        self._cache_lists.clear()
        self._cache_servers.clear()
        log.debug("Cleared metadata cache")

    async def close(self) -> None:
        """Close the HTTP client."""
        await self.client.aclose()
        log.debug("Closed HTTP client")

    async def __aenter__(self) -> Self:
        return self

    async def __aexit__(self, *args: object) -> None:
        await self.close()

clear_cache

clear_cache() -> None

Clear the metadata cache.

Source code in src/llmling_agent/mcp_server/registries/official_registry_client.py

def clear_cache(self) -> None:
    """Clear the metadata cache."""
    self._cache_lists.clear()
    self._cache_servers.clear()
    log.debug("Cleared metadata cache")

close async

close() -> None

Close the HTTP client.

Source code in src/llmling_agent/mcp_server/registries/official_registry_client.py

async def close(self) -> None:
    """Close the HTTP client."""
    await self.client.aclose()
    log.debug("Closed HTTP client")

get_server async

get_server(server_id: str) -> RegistryServer

Get full server details including packages.

Source code in src/llmling_agent/mcp_server/registries/official_registry_client.py

async def get_server(self, server_id: str) -> RegistryServer:
    """Get full server details including packages."""
    cache_key = f"server:{server_id}"

    # Check cache first
    if cache_key in self._cache_servers:
        cached_entry = self._cache_servers[cache_key]
        if time.time() - cached_entry.timestamp < CACHE_TTL:
            log.debug("Using cached server details for %s", server_id)
            return cached_entry.server

    try:
        log.info("Fetching server details for %s", server_id)

        # Get all wrappers to access metadata
        response = await self.client.get(f"{self.base_url}/v0/servers")
        response.raise_for_status()
        data = response.json()
        response_data = RegistryListResponse(**data)

        # Find server by name
        target_wrapper = None
        for wrapper in response_data.servers:
            if wrapper.server.name == server_id:
                target_wrapper = wrapper
                break

        if not target_wrapper:
            msg = f"Server {server_id!r} not found in registry"
            raise MCPRegistryError(msg)

        # Get the UUID from metadata
        server_uuid = target_wrapper.meta.get(
            "io.modelcontextprotocol.registry/official", {}
        ).get("id")

        if not server_uuid:
            msg = f"No UUID found for server {server_id!r}"
            raise MCPRegistryError(msg)

        # Now fetch the full server details using UUID
        response = await self.client.get(f"{self.base_url}/v0/servers/{server_uuid}")
        response.raise_for_status()

        server_data = response.json()
        server = RegistryServer(**server_data)

        # Cache the result
        self._cache_servers[cache_key] = GetServerCacheEntry(
            server=server, timestamp=time.time()
        )
        log.info("Successfully fetched server details for %s", server_id)

    except httpx.HTTPStatusError as e:
        if e.response.status_code == HTTP_NOT_FOUND:
            msg = f"Server {server_id!r} not found in registry"
            raise MCPRegistryError(msg) from e
        msg = f"Failed to get server details: {e}"
        raise MCPRegistryError(msg) from e
    except (httpx.HTTPError, ValueError, KeyError) as e:
        msg = f"Failed to get server details: {e}"
        raise MCPRegistryError(msg) from e
    else:
        return server

list_servers async

list_servers(search: str | None = None, status: str = 'active') -> list[RegistryServer]

List servers from registry with optional filtering.

Source code in src/llmling_agent/mcp_server/registries/official_registry_client.py

async def list_servers(
    self, search: str | None = None, status: str = "active"
) -> list[RegistryServer]:
    """List servers from registry with optional filtering."""
    cache_key = f"list_servers:{search}:{status}"

    # Check cache first
    if cache_key in self._cache_lists:
        cached_entry = self._cache_lists[cache_key]
        if time.time() - cached_entry.timestamp < CACHE_TTL:
            log.debug("Using cached server list")
            return cached_entry.servers

    try:
        log.info("Fetching server list from registry")
        response = await self.client.get(f"{self.base_url}/v0/servers")
        response.raise_for_status()
        data = response.json()
    except httpx.HTTPError as e:
        msg = f"Failed to list servers: {e}"
        raise MCPRegistryError(msg) from e
    else:
        response_data = RegistryListResponse(**data)
        wrappers = response_data.servers

        if status:  # Filter by status from metadata
            wrappers = [
                wrapper
                for wrapper in wrappers
                if wrapper.meta.get("io.modelcontextprotocol.registry/official", {}).get(
                    "status"
                )
                == status
            ]

        servers = [wrapper.server for wrapper in wrappers]
        if search:  # Filter by search term
            search_lower = search.lower()
            servers = [
                s
                for s in servers
                if search_lower in s.name.lower() or search_lower in s.description.lower()
            ]

        # Cache the result
        self._cache_lists[cache_key] = ListServersCacheEntry(
            servers=servers, timestamp=time.time()
        )
        log.info("Successfully fetched %d servers", len(servers))
        return servers

MCPRegistryError

Bases: Exception

Exception raised for MCP registry operations.

Source code in src/llmling_agent/mcp_server/registries/official_registry_client.py

class MCPRegistryError(Exception):
    """Exception raised for MCP registry operations."""

RegistryListResponse

Bases: Schema

Response from the registry list servers endpoint.

Source code in src/llmling_agent/mcp_server/registries/official_registry_client.py

class RegistryListResponse(Schema):
    """Response from the registry list servers endpoint."""

    servers: list[RegistryServerWrapper]
    """List of wrapped server entries."""

    metadata: dict[str, Any] = Field(default_factory=dict)
    """Response metadata."""

metadata class-attribute instance-attribute

metadata: dict[str, Any] = Field(default_factory=dict)

Response metadata.

servers instance-attribute

servers: list[RegistryServerWrapper]

List of wrapped server entries.

RegistryPackage

Bases: Schema

Package information for installing an MCP server.

Source code in src/llmling_agent/mcp_server/registries/official_registry_client.py

class RegistryPackage(Schema):
    """Package information for installing an MCP server."""

    registry_type: str = Field(alias="registryType")
    """Package registry type (npm, pypi, docker)."""

    identifier: str
    """Package identifier."""

    version: str | None = None
    """Package version."""

    transport: RegistryTransport
    """Transport configuration."""

    environment_variables: list[dict[str, Any]] = Field(
        default_factory=list, alias="environmentVariables"
    )
    """Environment variables."""

    package_arguments: list[dict[str, Any]] = Field(default_factory=list, alias="packageArguments")
    """Package arguments."""

    runtime_hint: str | None = Field(None, alias="runtimeHint")
    """Runtime hint."""

    registry_base_url: str | None = Field(None, alias="registryBaseUrl")
    """Registry base URL."""

    file_sha256: str | None = Field(None, alias="fileSha256")
    """File SHA256 hash."""

environment_variables class-attribute instance-attribute

environment_variables: list[dict[str, Any]] = Field(
    default_factory=list, alias="environmentVariables"
)

Environment variables.

file_sha256 class-attribute instance-attribute

file_sha256: str | None = Field(None, alias='fileSha256')

File SHA256 hash.

identifier instance-attribute

identifier: str

Package identifier.

package_arguments class-attribute instance-attribute

package_arguments: list[dict[str, Any]] = Field(default_factory=list, alias='packageArguments')

Package arguments.

registry_base_url class-attribute instance-attribute

registry_base_url: str | None = Field(None, alias='registryBaseUrl')

Registry base URL.

registry_type class-attribute instance-attribute

registry_type: str = Field(alias='registryType')

Package registry type (npm, pypi, docker).

runtime_hint class-attribute instance-attribute

runtime_hint: str | None = Field(None, alias='runtimeHint')

Runtime hint.

transport instance-attribute

transport: RegistryTransport

Transport configuration.

version class-attribute instance-attribute

version: str | None = None

Package version.

RegistryRemote

Bases: Schema

Remote endpoint configuration.

Source code in src/llmling_agent/mcp_server/registries/official_registry_client.py

class RegistryRemote(Schema):
    """Remote endpoint configuration."""

    type: str
    """Remote type (sse, streamable-http)."""

    url: str
    """Remote URL."""

    headers: list[dict[str, Any]] = Field(default_factory=list)
    """Request headers."""

headers class-attribute instance-attribute

headers: list[dict[str, Any]] = Field(default_factory=list)

Request headers.

type instance-attribute

type: str

Remote type (sse, streamable-http).

url instance-attribute

url: str

Remote URL.

RegistryRepository

Bases: Schema

Repository information for a registry server.

Source code in src/llmling_agent/mcp_server/registries/official_registry_client.py

class RegistryRepository(Schema):
    """Repository information for a registry server."""

    url: str
    source: str
    """Repository platform (e.g., 'github')."""

    @field_validator("url")
    @classmethod
    def validate_url(cls, v: str) -> str:
        """Allow empty URLs from registry."""
        return v

source instance-attribute

source: str

Repository platform (e.g., 'github').

validate_url classmethod

validate_url(v: str) -> str

Allow empty URLs from registry.

Source code in src/llmling_agent/mcp_server/registries/official_registry_client.py

@field_validator("url")
@classmethod
def validate_url(cls, v: str) -> str:
    """Allow empty URLs from registry."""
    return v

RegistryServer

Bases: Schema

MCP server entry from the registry.

Source code in src/llmling_agent/mcp_server/registries/official_registry_client.py

class RegistryServer(Schema):
    """MCP server entry from the registry."""

    name: ServiceName
    """Unique server identifier."""

    description: str
    """Server description."""

    version: str
    """Server version."""

    repository: RegistryRepository
    """Repository information."""

    packages: list[RegistryPackage] = Field(default_factory=list)
    """Available packages."""

    remotes: list[RegistryRemote] = Field(default_factory=list)
    """Remote endpoints."""

    schema_: str | None = Field(None, alias="$schema")
    """JSON schema URL."""

    def get_preferred_transport(self) -> TransportType:
        """Select optimal transport method based on availability and performance."""
        # Prefer local packages for better performance/security
        for package in self.packages:
            if package.registry_type in ["docker", "oci"]:  # OCI containers
                return "stdio"

        # Fallback to remote endpoints
        for remote in self.remotes:
            if remote.type == "sse":
                return "sse"
            if remote.type in ["streamable-http", "http"]:
                return "http"
            if remote.type == "websocket":
                return "websocket"

        # Provide helpful error message
        available_transports = []
        if self.packages:
            available_transports.extend([f"package:{pkg.registry_type}" for pkg in self.packages])
        if self.remotes:
            available_transports.extend([f"remote:{remote.type}" for remote in self.remotes])

        if available_transports:
            error_msg = (
                f"No supported transport for {self.name}. "
                f"Available: {available_transports}. "
                f"Supported: docker packages, sse/streamable-http/websocket remotes"
            )
        else:
            error_msg = (
                f"No transports available for {self.name}. Server metadata may be incomplete"
            )

        raise UnsupportedTransportError(error_msg)

description instance-attribute

description: str

Server description.

name instance-attribute

name: ServiceName

Unique server identifier.

packages class-attribute instance-attribute

packages: list[RegistryPackage] = Field(default_factory=list)

Available packages.

remotes class-attribute instance-attribute

remotes: list[RegistryRemote] = Field(default_factory=list)

Remote endpoints.

repository instance-attribute

repository: RegistryRepository

Repository information.

schema_ class-attribute instance-attribute

schema_: str | None = Field(None, alias='$schema')

JSON schema URL.

version instance-attribute

version: str

Server version.

get_preferred_transport

get_preferred_transport() -> TransportType

Select optimal transport method based on availability and performance.

Source code in src/llmling_agent/mcp_server/registries/official_registry_client.py

def get_preferred_transport(self) -> TransportType:
    """Select optimal transport method based on availability and performance."""
    # Prefer local packages for better performance/security
    for package in self.packages:
        if package.registry_type in ["docker", "oci"]:  # OCI containers
            return "stdio"

    # Fallback to remote endpoints
    for remote in self.remotes:
        if remote.type == "sse":
            return "sse"
        if remote.type in ["streamable-http", "http"]:
            return "http"
        if remote.type == "websocket":
            return "websocket"

    # Provide helpful error message
    available_transports = []
    if self.packages:
        available_transports.extend([f"package:{pkg.registry_type}" for pkg in self.packages])
    if self.remotes:
        available_transports.extend([f"remote:{remote.type}" for remote in self.remotes])

    if available_transports:
        error_msg = (
            f"No supported transport for {self.name}. "
            f"Available: {available_transports}. "
            f"Supported: docker packages, sse/streamable-http/websocket remotes"
        )
    else:
        error_msg = (
            f"No transports available for {self.name}. Server metadata may be incomplete"
        )

    raise UnsupportedTransportError(error_msg)

RegistryServerWrapper

Bases: Schema

Wrapper for server data from the official registry API.

Source code in src/llmling_agent/mcp_server/registries/official_registry_client.py

class RegistryServerWrapper(Schema):
    """Wrapper for server data from the official registry API."""

    server: RegistryServer
    """The actual server data."""

    meta: dict[str, Any] = Field(default_factory=dict, alias="_meta")
    """Registry metadata."""

meta class-attribute instance-attribute

meta: dict[str, Any] = Field(default_factory=dict, alias='_meta')

Registry metadata.

server instance-attribute

server: RegistryServer

The actual server data.

RegistryTransport

Bases: Schema

Transport configuration for a package.

Source code in src/llmling_agent/mcp_server/registries/official_registry_client.py

class RegistryTransport(Schema):
    """Transport configuration for a package."""

    type: str
    """Transport type (stdio, sse, streamable-http)."""

    url: str | None = None
    """URL for HTTP transports."""

type instance-attribute

type: str

Transport type (stdio, sse, streamable-http).

url class-attribute instance-attribute

url: str | None = None

URL for HTTP transports.

UnsupportedTransportError

Bases: Exception

Raised when no supported transport is available.

Source code in src/llmling_agent/mcp_server/registries/official_registry_client.py

class UnsupportedTransportError(Exception):
    """Raised when no supported transport is available."""

main async

main() -> None

Test the MCP registry client with caching and transport resolution.

Source code in src/llmling_agent/mcp_server/registries/official_registry_client.py

async def main() -> None:
    """Test the MCP registry client with caching and transport resolution."""
    async with MCPRegistryClient() as client:
        print("Listing servers from official registry...")
        servers = await client.list_servers()
        print(f"Found {len(servers)} servers")

        # Test transport resolution for first few servers
        for server in servers[:3]:
            print(f"\n=== {server.name} ===")
            try:
                transport = server.get_preferred_transport()
                print(f"Preferred transport: {transport}")

                # Show available transports
                if server.packages:
                    print(f"Packages: {[p.registry_type for p in server.packages]}")
                if server.remotes:
                    print(f"Remotes: {[r.type for r in server.remotes]}")

            except UnsupportedTransportError as e:
                print(f"Transport error: {e}")

            print(server)