Skip to content

MCP Server

lcp.mcp_server

MCP server that exposes LCP manifest data to AI agents.

DEFAULT_MAX_RESPONSE_BYTES module-attribute

DEFAULT_MAX_RESPONSE_BYTES = 25000

Default byte budget for any list-returning tool payload (spec D6).

Calibrated against polars 1.42.1: DataFrame's 159 member summaries alone are ~30 KB, so heavy classes are expected to truncate; 25 KB is ~6k tokens.

LCPIndex

In-memory index of LCP document for fast lookups.

Source code in src/lcp/mcp_server.py
class LCPIndex:
    """In-memory index of LCP document for fast lookups."""

    def __init__(self, doc: LCPDocument):
        self.doc = doc
        self.symbols_by_id: dict[str, Symbol] = doc.symbols
        self.symbols_by_module: dict[str, list[str]] = defaultdict(list)
        self.symbols_by_kind: dict[str, list[str]] = defaultdict(list)
        self.class_members: dict[str, list[str]] = defaultdict(list)
        self.classes_by_name: dict[str, list[str]] = defaultdict(list)
        self.modules: set[str] = set()
        self.alias_to_canonical: dict[str, str] = {}
        self.preferred_alias: dict[str, str] = {}

        self._build_indexes()

    def _build_indexes(self) -> None:
        """Build lookup indexes from the LCP document."""
        for symbol_id, symbol in self.symbols_by_id.items():
            # Index by module
            if symbol.module:
                self.symbols_by_module[symbol.module].append(symbol_id)
                self.modules.add(symbol.module)

            # Index by kind
            self.symbols_by_kind[symbol.kind.value].append(symbol_id)

            # Index class members (symbols with # in ID belong to a class)
            if "#" in symbol_id:
                class_id = symbol_id.split("#")[0]
                self.class_members[class_id].append(symbol_id)

            # Index classes by bare name for exact return-type resolution
            if symbol.kind == SymbolKind.CLASS:
                self.classes_by_name[_symbol_name(symbol_id)].append(symbol_id)

        for ids in self.classes_by_name.values():
            ids.sort()

        # Alias indexes (spec D10): alias ids resolve to the canonical
        # entry, and class aliases expand to member ids at build time —
        # without materializing duplicate entries in the manifest.
        for symbol_id, symbol in self.symbols_by_id.items():
            aliases = [
                a
                for a in (symbol.aliases or [])
                if a not in self.symbols_by_id
            ]
            if not aliases:
                continue
            for alias in aliases:
                self.alias_to_canonical.setdefault(alias, symbol_id)
            preferred = min(aliases, key=_alias_rank)
            self.preferred_alias[symbol_id] = preferred
            if symbol.kind == SymbolKind.CLASS:
                for member_id in self.class_members.get(symbol_id, []):
                    entity = member_id.split("#", 1)[1]
                    for alias in aliases:
                        self.alias_to_canonical.setdefault(
                            f"{alias}#{entity}", member_id
                        )
                    self.preferred_alias[member_id] = f"{preferred}#{entity}"

    def resolve_id(self, symbol_id: str) -> tuple[str | None, Symbol | None]:
        """Resolve a possibly-aliased id to ``(canonical_id, symbol)``.

        Returns ``(None, None)`` when the id matches neither a canonical
        entry nor a known alias.
        """
        symbol = self.symbols_by_id.get(symbol_id)
        if symbol is not None:
            return symbol_id, symbol
        canonical = self.alias_to_canonical.get(symbol_id)
        if canonical is not None:
            return canonical, self.symbols_by_id[canonical]
        return None, None

resolve_id

resolve_id(symbol_id: str) -> tuple[str | None, Symbol | None]

Resolve a possibly-aliased id to (canonical_id, symbol).

Returns (None, None) when the id matches neither a canonical entry nor a known alias.

Source code in src/lcp/mcp_server.py
def resolve_id(self, symbol_id: str) -> tuple[str | None, Symbol | None]:
    """Resolve a possibly-aliased id to ``(canonical_id, symbol)``.

    Returns ``(None, None)`` when the id matches neither a canonical
    entry nor a known alias.
    """
    symbol = self.symbols_by_id.get(symbol_id)
    if symbol is not None:
        return symbol_id, symbol
    canonical = self.alias_to_canonical.get(symbol_id)
    if canonical is not None:
        return canonical, self.symbols_by_id[canonical]
    return None, None

MultiLibraryIndex

Registry of loaded LCPIndex instances for the MCP server.

Holds one :class:LCPIndex per loaded library plus the source it was resolved from. There is deliberately no implicit default library: with one library loaded :meth:resolve returns it for a None argument, with several it returns an ambiguous_library error (spec D7).

Source code in src/lcp/mcp_server.py
class MultiLibraryIndex:
    """Registry of loaded LCPIndex instances for the MCP server.

    Holds one :class:`LCPIndex` per loaded library plus the source it was
    resolved from. There is deliberately **no** implicit default library:
    with one library loaded :meth:`resolve` returns it for a ``None``
    argument, with several it returns an ``ambiguous_library`` error
    (spec D7).
    """

    def __init__(self) -> None:
        self._entries: dict[str, tuple[LCPIndex, str]] = {}

    def add(self, name: str, index: LCPIndex, source: str = "scan") -> None:
        """Register (or replace) a library index.

        Args:
            name: Library name used as the lookup key.
            index: Built index for the library's manifest.
            source: Where the manifest came from (``"cache"``, ``"scan"``,
                ``"registry"``, or ``"manifest"`` for a pre-loaded file).
        """
        self._entries[name] = (index, source)

    def get(self, name: str) -> LCPIndex | None:
        """Return the index registered under *name*, or None."""
        entry = self._entries.get(name)
        return entry[0] if entry else None

    def source(self, name: str) -> str | None:
        """Return the resolution source recorded for *name*, or None."""
        entry = self._entries.get(name)
        return entry[1] if entry else None

    def names(self) -> list[str]:
        """Return the sorted names of all loaded libraries."""
        return sorted(self._entries)

    def resolve(
        self, library: str | None
    ) -> tuple[str | None, LCPIndex | None, dict[str, Any] | None]:
        """Resolve a tool's ``library`` argument to an index (spec D7).

        Args:
            library: Explicit library name, or None.

        Returns:
            ``(name, index, None)`` on success, ``(None, None, error_dict)``
            on failure — the error dict follows the D5 shape.
        """
        if library is not None:
            entry = self._entries.get(library)
            if entry is not None:
                return library, entry[0], None
            return None, None, _error(
                "library_not_loaded",
                f"Library '{library}' is not loaded.",
                hint=f"Call resolve_library('{library}') first.",
                loaded_libraries=self.names(),
            )
        if not self._entries:
            return None, None, _error(
                "library_not_loaded",
                "No library is loaded.",
                hint="Call resolve_library(<package name>) first.",
            )
        if len(self._entries) == 1:
            name = next(iter(self._entries))
            return name, self._entries[name][0], None
        return None, None, _error(
            "ambiguous_library",
            "Multiple libraries are loaded; pass library=<name>.",
            hint="Pick one of loaded_libraries and retry with library=<name>.",
            loaded_libraries=self.names(),
        )

    def list_libraries(self) -> list[dict[str, Any]]:
        """Return summary info for all loaded libraries."""
        result = []
        for name in self.names():
            idx, source = self._entries[name]
            lib = idx.doc.manifest.library
            result.append(
                {
                    "name": name,
                    "version": lib.version,
                    "language": lib.language,
                    "symbol_count": len(idx.symbols_by_id),
                    "source": source,
                }
            )
        return result

    def __contains__(self, name: str) -> bool:
        return name in self._entries

add

add(name: str, index: LCPIndex, source: str = 'scan') -> None

Register (or replace) a library index.

Parameters:

Name Type Description Default
name str

Library name used as the lookup key.

required
index LCPIndex

Built index for the library's manifest.

required
source str

Where the manifest came from ("cache", "scan", "registry", or "manifest" for a pre-loaded file).

'scan'
Source code in src/lcp/mcp_server.py
def add(self, name: str, index: LCPIndex, source: str = "scan") -> None:
    """Register (or replace) a library index.

    Args:
        name: Library name used as the lookup key.
        index: Built index for the library's manifest.
        source: Where the manifest came from (``"cache"``, ``"scan"``,
            ``"registry"``, or ``"manifest"`` for a pre-loaded file).
    """
    self._entries[name] = (index, source)

get

get(name: str) -> LCPIndex | None

Return the index registered under name, or None.

Source code in src/lcp/mcp_server.py
def get(self, name: str) -> LCPIndex | None:
    """Return the index registered under *name*, or None."""
    entry = self._entries.get(name)
    return entry[0] if entry else None

source

source(name: str) -> str | None

Return the resolution source recorded for name, or None.

Source code in src/lcp/mcp_server.py
def source(self, name: str) -> str | None:
    """Return the resolution source recorded for *name*, or None."""
    entry = self._entries.get(name)
    return entry[1] if entry else None

names

names() -> list[str]

Return the sorted names of all loaded libraries.

Source code in src/lcp/mcp_server.py
def names(self) -> list[str]:
    """Return the sorted names of all loaded libraries."""
    return sorted(self._entries)

resolve

resolve(library: str | None) -> tuple[str | None, LCPIndex | None, dict[str, Any] | None]

Resolve a tool's library argument to an index (spec D7).

Parameters:

Name Type Description Default
library str | None

Explicit library name, or None.

required

Returns:

Type Description
str | None

(name, index, None) on success, (None, None, error_dict)

LCPIndex | None

on failure — the error dict follows the D5 shape.

Source code in src/lcp/mcp_server.py
def resolve(
    self, library: str | None
) -> tuple[str | None, LCPIndex | None, dict[str, Any] | None]:
    """Resolve a tool's ``library`` argument to an index (spec D7).

    Args:
        library: Explicit library name, or None.

    Returns:
        ``(name, index, None)`` on success, ``(None, None, error_dict)``
        on failure — the error dict follows the D5 shape.
    """
    if library is not None:
        entry = self._entries.get(library)
        if entry is not None:
            return library, entry[0], None
        return None, None, _error(
            "library_not_loaded",
            f"Library '{library}' is not loaded.",
            hint=f"Call resolve_library('{library}') first.",
            loaded_libraries=self.names(),
        )
    if not self._entries:
        return None, None, _error(
            "library_not_loaded",
            "No library is loaded.",
            hint="Call resolve_library(<package name>) first.",
        )
    if len(self._entries) == 1:
        name = next(iter(self._entries))
        return name, self._entries[name][0], None
    return None, None, _error(
        "ambiguous_library",
        "Multiple libraries are loaded; pass library=<name>.",
        hint="Pick one of loaded_libraries and retry with library=<name>.",
        loaded_libraries=self.names(),
    )

list_libraries

list_libraries() -> list[dict[str, Any]]

Return summary info for all loaded libraries.

Source code in src/lcp/mcp_server.py
def list_libraries(self) -> list[dict[str, Any]]:
    """Return summary info for all loaded libraries."""
    result = []
    for name in self.names():
        idx, source = self._entries[name]
        lib = idx.doc.manifest.library
        result.append(
            {
                "name": name,
                "version": lib.version,
                "language": lib.language,
                "symbol_count": len(idx.symbols_by_id),
                "source": source,
            }
        )
    return result

LCPServer dataclass

A configured LCP MCP server plus direct access to its internals.

Attributes:

Name Type Description
mcp FastMCP

The underlying FastMCP server (use :meth:run to serve stdio).

index MultiLibraryIndex

The registry of loaded library indexes.

tools dict[str, Callable[..., Any]]

Raw tool callables by name, for in-process invocation (tests, preload) without the MCP protocol.

Source code in src/lcp/mcp_server.py
@dataclass
class LCPServer:
    """A configured LCP MCP server plus direct access to its internals.

    Attributes:
        mcp: The underlying FastMCP server (use :meth:`run` to serve stdio).
        index: The registry of loaded library indexes.
        tools: Raw tool callables by name, for in-process invocation
            (tests, preload) without the MCP protocol.
    """

    mcp: FastMCP
    index: MultiLibraryIndex
    tools: dict[str, Callable[..., Any]]

    def run(self) -> None:
        """Run the MCP server on stdio transport (blocks)."""
        self.mcp.run()

run

run() -> None

Run the MCP server on stdio transport (blocks).

Source code in src/lcp/mcp_server.py
def run(self) -> None:
    """Run the MCP server on stdio transport (blocks)."""
    self.mcp.run()

load_lcp_document

load_lcp_document(path: str | Path) -> LCPDocument

Load and validate an LCP document from a file.

Supports both plain .lcp.json files and gzip-compressed .lcp.json.gz files, detected transparently by file extension.

Source code in src/lcp/mcp_server.py
def load_lcp_document(path: str | Path) -> LCPDocument:
    """Load and validate an LCP document from a file.

    Supports both plain ``.lcp.json`` files and gzip-compressed
    ``.lcp.json.gz`` files, detected transparently by file extension.
    """
    import gzip as _gzip

    path = Path(path)
    if not path.exists():
        raise FileNotFoundError(f"LCP file not found: {path}")

    if path.suffix == ".gz":
        with _gzip.open(path, "rb") as f:
            data = json.loads(f.read())
    else:
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)

    return LCPDocument.model_validate(data)

resolve_library_document

resolve_library_document(name: str, cache_dir: Path = _DEFAULT_CACHE_DIR, no_cache: bool = False, registry_url: str | None = None, version: str | None = None, scan_mode: str = 'subprocess', scan_python: str | None = None, scan_timeout: float = DEFAULT_SCAN_TIMEOUT) -> tuple[ScanResult, str]

Resolve an LCP document for name using the standard resolution order.

Resolution order
  1. Local cache (~/.lcp/cache/{name}/{version}.lcp.json)
  2. Live scan (subprocess by default; see scan_mode)
  3. Registry (HTTP GET from registry_url if provided)
  4. Error

Registry manifests are fetched from the path {registry_url}/manifests/python/{name}/{version}.lcp.json. When the installed version is unknown, "latest" is used as the version segment so registries can expose a canonical latest entry.

Parameters:

Name Type Description Default
name str

Python package name to resolve.

required
cache_dir Path

Cache root directory (default: ~/.lcp/cache/).

_DEFAULT_CACHE_DIR
no_cache bool

Skip cache read/write entirely.

False
registry_url str | None

Optional base URL of an LCP registry to try when local scanning fails. The default official registry is at https://raw.githubusercontent.com/zazza123/lcp-registry/refs/heads/main.

None
version str | None

Optional exact version to prefer; overrides the installed version for cache lookup and registry fetch. A live scan always returns the installed version regardless.

None
scan_mode str

"subprocess" (default) runs the live scan in a child interpreter — import-time crashes and heavy imports stay out of the server process; "inprocess" imports the package into this process (for environments where spawning is restricted).

'subprocess'
scan_python str | None

Interpreter whose environment the live scan reads (default: this process's interpreter). This is how packages installed in a different venv get resolved.

None
scan_timeout float

Seconds before a subprocess scan is killed.

DEFAULT_SCAN_TIMEOUT

Returns:

Type Description
ScanResult

Tuple of (ScanResult, source) where source is "cache",

str

"scan", or "registry". Only a "scan" result can carry

tuple[ScanResult, str]

unresolved_reexports; cache and registry hits carry an empty

tuple[ScanResult, str]

list, because the diagnostic is not persisted in the manifest.

Raises:

Type Description
ImportError

If the package cannot be resolved via any available source (cache, scan, or registry).

Source code in src/lcp/mcp_server.py
def resolve_library_document(
    name: str,
    cache_dir: Path = _DEFAULT_CACHE_DIR,
    no_cache: bool = False,
    registry_url: str | None = None,
    version: str | None = None,
    scan_mode: str = "subprocess",
    scan_python: str | None = None,
    scan_timeout: float = DEFAULT_SCAN_TIMEOUT,
) -> tuple[ScanResult, str]:
    """Resolve an LCP document for *name* using the standard resolution order.

    Resolution order:
      1. Local cache  (~/.lcp/cache/{name}/{version}.lcp.json)
      2. Live scan    (subprocess by default; see *scan_mode*)
      3. Registry     (HTTP GET from *registry_url* if provided)
      4. Error

    Registry manifests are fetched from the path
    ``{registry_url}/manifests/python/{name}/{version}.lcp.json``.
    When the installed version is unknown, ``"latest"`` is used as the
    version segment so registries can expose a canonical latest entry.

    Args:
        name: Python package name to resolve.
        cache_dir: Cache root directory (default: ~/.lcp/cache/).
        no_cache: Skip cache read/write entirely.
        registry_url: Optional base URL of an LCP registry to try when local
            scanning fails.  The default official registry is at
            ``https://raw.githubusercontent.com/zazza123/lcp-registry/refs/heads/main``.
        version: Optional exact version to prefer; overrides the installed
            version for cache lookup and registry fetch.  A live scan always
            returns the installed version regardless.
        scan_mode: ``"subprocess"`` (default) runs the live scan in a child
            interpreter — import-time crashes and heavy imports stay out of
            the server process; ``"inprocess"`` imports the package into
            this process (for environments where spawning is restricted).
        scan_python: Interpreter whose environment the live scan reads
            (default: this process's interpreter). This is how packages
            installed in a different venv get resolved.
        scan_timeout: Seconds before a subprocess scan is killed.

    Returns:
        Tuple of (ScanResult, source) where source is ``"cache"``,
        ``"scan"``, or ``"registry"``. Only a ``"scan"`` result can carry
        ``unresolved_reexports``; cache and registry hits carry an empty
        list, because the diagnostic is not persisted in the manifest.

    Raises:
        ImportError: If the package cannot be resolved via any available
            source (cache, scan, or registry).
    """
    # Resolve the installed version once; used for both cache lookup and
    # registry fetch. An explicit *version* takes precedence over it.
    installed_ver = _installed_version(name)
    lookup_ver = version or installed_ver

    # 1. Cache lookup
    if not no_cache:
        if lookup_ver:
            # Exact-version match (requested version wins over installed)
            cached = _load_from_cache(cache_dir, name, lookup_ver)
            if cached is not None:
                return ScanResult(document=cached), "cache"
        else:
            # No version to pin: return any cached entry for this package
            cached = _find_any_cached(cache_dir, name)
            if cached is not None:
                return ScanResult(document=cached), "cache"

    # 2. Live scan
    scan_error: Exception | None = None
    try:
        result = _scan_live(name, scan_mode, scan_python, scan_timeout)
        if not no_cache:
            try:
                _save_to_cache(cache_dir, result.document)
            except Exception:
                pass  # cache write failure is non-fatal
        return result, "scan"
    except Exception as exc:
        scan_error = exc

    # 3. Registry fallback
    if registry_url:
        try:
            doc = _fetch_from_registry(name, registry_url, version=lookup_ver)
            if not no_cache:
                try:
                    _save_to_cache(cache_dir, doc)
                except Exception:
                    pass  # cache write failure is non-fatal
            return ScanResult(document=doc), "registry"
        except ImportError:
            pass  # fall through to final error

    scan_interpreter = scan_python or sys.executable
    if isinstance(scan_error, ScanImportError):
        if scan_python:
            remedy = (
                "Check that the package is installed in that environment, "
                "or fix 'scan_python' in .lcp-config.json"
            )
        else:
            remedy = (
                "It may be installed in a different environment — point the "
                "server at that env via .lcp-config.json ('scan_python' or 'python')"
            )
        reason = (
            f"'{name}' is not importable by the scan interpreter "
            f"({scan_interpreter}). {remedy}; the distribution name may also "
            f"differ from the import path (e.g. import 'google.adk' is "
            f"provided by 'pip install google-adk')."
        )
    elif isinstance(
        scan_error, (ScanTimeoutError, ScanInterpreterNotFoundError, ScanSpawnError)
    ):
        # The runner's messages are already agent-facing and actionable.
        reason = str(scan_error)
    elif installed_ver:
        # The package is installed in this environment but scanning failed.
        reason = (
            f"'{name}' is installed (version {installed_ver}) in this environment "
            f"but the scan failed: {type(scan_error).__name__}: {scan_error}"
        )
    elif isinstance(scan_error, ImportError):
        # In-process scan: could not import with the interpreter running lcp.
        reason = (
            f"'{name}' is not importable by the Python interpreter running lcp "
            f"({sys.executable}). It may be installed in a different environment "
            f"(point the plugin at that env via .lcp-config.json), or the distribution "
            f"name may differ from the import path "
            f"(e.g. import 'google.adk' is provided by 'pip install google-adk')."
        )
    else:
        reason = (
            f"could not resolve '{name}': {type(scan_error).__name__}: {scan_error}"
        )

    raise ImportError(
        f"Cannot resolve library '{name}': {reason}"
        + (f" (registry fetch also failed: {registry_url})" if registry_url else "")
    ) from scan_error

create_universal_server

create_universal_server(name: str = 'lcp-universal', cache_dir: Path | str | None = None, no_cache: bool = False, registry_url: str | None = None, expose: list[str] | None = None, preload: list[str] | None = None, max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES, scan_mode: str = 'subprocess', scan_python: str | None = None, scan_timeout: float = DEFAULT_SCAN_TIMEOUT) -> LCPServer

Create a universal MCP server that resolves any installed Python library.

The server exposes exactly four tools — resolve_library, search, get_symbol, get_overview — and carries adoption-focused instructions so agents verify APIs before writing code.

Parameters:

Name Type Description Default
name str

Server name shown to MCP clients.

'lcp-universal'
cache_dir Path | str | None

Root directory for cached manifests (default ~/.lcp/cache/).

None
no_cache bool

Disable reading from and writing to the cache.

False
registry_url str | None

Optional LCP registry URL used when local scanning fails.

None
expose list[str] | None

Optional allow-list of package names resolve_library may load.

None
preload list[str] | None

Package names to resolve eagerly at startup.

None
max_response_bytes int

Byte budget for list-returning tool responses.

DEFAULT_MAX_RESPONSE_BYTES
scan_mode str

Live-scan strategy: "subprocess" (default) isolates package imports in a child process; "inprocess" imports into the server process.

'subprocess'
scan_python str | None

Interpreter whose environment live scans read (default: the interpreter running the server).

None
scan_timeout float

Seconds before a subprocess scan is killed.

DEFAULT_SCAN_TIMEOUT

Returns:

Name Type Description
An LCPServer

class:LCPServer bundling the FastMCP instance, the library

LCPServer

index, and the raw tool callables.

Source code in src/lcp/mcp_server.py
def create_universal_server(
    name: str = "lcp-universal",
    cache_dir: Path | str | None = None,
    no_cache: bool = False,
    registry_url: str | None = None,
    expose: list[str] | None = None,
    preload: list[str] | None = None,
    max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES,
    scan_mode: str = "subprocess",
    scan_python: str | None = None,
    scan_timeout: float = DEFAULT_SCAN_TIMEOUT,
) -> LCPServer:
    """Create a universal MCP server that resolves any installed Python library.

    The server exposes exactly four tools — ``resolve_library``, ``search``,
    ``get_symbol``, ``get_overview`` — and carries adoption-focused
    ``instructions`` so agents verify APIs before writing code.

    Args:
        name: Server name shown to MCP clients.
        cache_dir: Root directory for cached manifests (default ~/.lcp/cache/).
        no_cache: Disable reading from and writing to the cache.
        registry_url: Optional LCP registry URL used when local scanning fails.
        expose: Optional allow-list of package names resolve_library may load.
        preload: Package names to resolve eagerly at startup.
        max_response_bytes: Byte budget for list-returning tool responses.
        scan_mode: Live-scan strategy: ``"subprocess"`` (default) isolates
            package imports in a child process; ``"inprocess"`` imports into
            the server process.
        scan_python: Interpreter whose environment live scans read
            (default: the interpreter running the server).
        scan_timeout: Seconds before a subprocess scan is killed.

    Returns:
        An :class:`LCPServer` bundling the FastMCP instance, the library
        index, and the raw tool callables.
    """
    resolved_cache_dir = Path(cache_dir) if cache_dir else _DEFAULT_CACHE_DIR
    libraries = MultiLibraryIndex()
    mcp = FastMCP(name, instructions=SERVER_INSTRUCTIONS)
    # An expose list that is non-empty but blank after stripping must fail
    # closed (block everything), not fall back to allow-all.
    allow: set[str] | None = None
    if expose:
        allow = {n.strip() for n in expose if n.strip()}

    tools = _register_tools(
        mcp,
        libraries,
        cache_dir=resolved_cache_dir,
        no_cache=no_cache,
        registry_url=registry_url,
        allow=allow,
        max_response_bytes=max_response_bytes,
        scan_mode=scan_mode,
        scan_python=scan_python,
        scan_timeout=scan_timeout,
    )

    for pkg in preload or []:
        try:
            result = tools["resolve_library"](pkg)
            if "error" in result:
                raise RuntimeError(result["error"]["message"])
        except Exception as exc:
            print(
                f"Warning: failed to preload package '{pkg}': {exc}",
                file=sys.stderr,
            )

    return LCPServer(mcp=mcp, index=libraries, tools=tools)

create_server

create_server(manifest_path: str | Path, name: str | None = None) -> LCPServer

Create an MCP server pre-loaded with one LCP manifest.

.. deprecated:: lcp serve / create_server are deprecated; use lcp serve-all --expose <package> / :func:create_universal_server. This wrapper builds the same universal server with the manifest pre-loaded and resolution locked to its library.

Parameters:

Name Type Description Default
manifest_path str | Path

Path to the .lcp.json file.

required
name str | None

Server name (default: lcp-{library-name}).

None

Returns:

Name Type Description
Configured LCPServer

class:LCPServer instance.

Source code in src/lcp/mcp_server.py
def create_server(
    manifest_path: str | Path,
    name: str | None = None,
) -> LCPServer:
    """Create an MCP server pre-loaded with one LCP manifest.

    .. deprecated::
        ``lcp serve`` / ``create_server`` are deprecated; use
        ``lcp serve-all --expose <package>`` / :func:`create_universal_server`.
        This wrapper builds the same universal server with the manifest
        pre-loaded and resolution locked to its library.

    Args:
        manifest_path: Path to the ``.lcp.json`` file.
        name: Server name (default: ``lcp-{library-name}``).

    Returns:
        Configured :class:`LCPServer` instance.
    """
    warnings.warn(
        "create_server()/'lcp serve' are deprecated; use "
        "create_universal_server()/'lcp serve-all --expose <package>'.",
        DeprecationWarning,
        stacklevel=2,
    )
    doc = load_lcp_document(manifest_path)
    lib_name = doc.manifest.library.name
    server = create_universal_server(
        name=name or f"lcp-{lib_name}", expose=[lib_name]
    )
    server.index.add(lib_name, LCPIndex(doc), source="manifest")
    return server

run_server

run_server(manifest_path: str | Path, name: str | None = None) -> None

Create and run a (deprecated) single-manifest MCP server.

Parameters:

Name Type Description Default
manifest_path str | Path

Path to the .lcp.json file.

required
name str | None

Server name (default: lcp-{library-name}).

None
Source code in src/lcp/mcp_server.py
def run_server(manifest_path: str | Path, name: str | None = None) -> None:
    """Create and run a (deprecated) single-manifest MCP server.

    Args:
        manifest_path: Path to the ``.lcp.json`` file.
        name: Server name (default: ``lcp-{library-name}``).
    """
    create_server(manifest_path, name=name).run()

run_universal_server

run_universal_server(name: str = 'lcp-universal', cache_dir: Path | str | None = None, no_cache: bool = False, registry_url: str | None = None, expose: list[str] | None = None, preload: list[str] | None = None, max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES, scan_mode: str = 'subprocess', scan_python: str | None = None, scan_timeout: float = DEFAULT_SCAN_TIMEOUT) -> None

Create and run a universal MCP server that resolves any installed Python library.

Parameters:

Name Type Description Default
name str

Server name (default: lcp-universal).

'lcp-universal'
cache_dir Path | str | None

Root directory for cached manifests (default: ~/.lcp/cache/).

None
no_cache bool

Disable reading from and writing to the cache.

False
registry_url str | None

Optional base URL of an LCP registry used as a fallback when local scanning fails.

None
expose list[str] | None

Optional allow-list of package names for resolve_library. When None or empty, all packages are allowed.

None
preload list[str] | None

Package names to resolve eagerly at startup.

None
max_response_bytes int

Byte budget for list-returning tool responses.

DEFAULT_MAX_RESPONSE_BYTES
scan_mode str

Live-scan strategy: "subprocess" (default) or "inprocess".

'subprocess'
scan_python str | None

Interpreter whose environment live scans read (default: the interpreter running the server).

None
scan_timeout float

Seconds before a subprocess scan is killed.

DEFAULT_SCAN_TIMEOUT
Source code in src/lcp/mcp_server.py
def run_universal_server(
    name: str = "lcp-universal",
    cache_dir: Path | str | None = None,
    no_cache: bool = False,
    registry_url: str | None = None,
    expose: list[str] | None = None,
    preload: list[str] | None = None,
    max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES,
    scan_mode: str = "subprocess",
    scan_python: str | None = None,
    scan_timeout: float = DEFAULT_SCAN_TIMEOUT,
) -> None:
    """Create and run a universal MCP server that resolves any installed Python library.

    Args:
        name: Server name (default: lcp-universal).
        cache_dir: Root directory for cached manifests (default: ~/.lcp/cache/).
        no_cache: Disable reading from and writing to the cache.
        registry_url: Optional base URL of an LCP registry used as a fallback
            when local scanning fails.
        expose: Optional allow-list of package names for ``resolve_library``.
            When ``None`` or empty, all packages are allowed.
        preload: Package names to resolve eagerly at startup.
        max_response_bytes: Byte budget for list-returning tool responses.
        scan_mode: Live-scan strategy: ``"subprocess"`` (default) or
            ``"inprocess"``.
        scan_python: Interpreter whose environment live scans read
            (default: the interpreter running the server).
        scan_timeout: Seconds before a subprocess scan is killed.
    """
    server = create_universal_server(
        name=name,
        cache_dir=cache_dir,
        no_cache=no_cache,
        registry_url=registry_url,
        expose=expose,
        preload=preload,
        max_response_bytes=max_response_bytes,
        scan_mode=scan_mode,
        scan_python=scan_python,
        scan_timeout=scan_timeout,
    )
    server.run()