server

Class info

Classes

Name	Children	Inherits
ServerBridge llmling_agent.server Base class for AgentPool bridge servers.		ABC

🛈 DocStrings

Base classes for protocol bridge servers.

ServerBridge

Bases: ABC

Base class for AgentPool bridge servers.

Provides common lifecycle management, context manager protocol, and running state handling for servers that bridge llmling agents to external protocols (ACP, OpenAI API, MCP, etc.).

Source code in src/llmling_agent/server.py

class ServerBridge(abc.ABC):
    """Base class for AgentPool bridge servers.

    Provides common lifecycle management, context manager protocol,
    and running state handling for servers that bridge llmling agents
    to external protocols (ACP, OpenAI API, MCP, etc.).
    """

    def __init__(
        self,
        pool: AgentPool[Any],
        **kwargs: Any,
    ):
        """Initialize the server bridge.

        Args:
            pool: Agent pool to expose via the protocol
            **kwargs: Additional configuration options
        """
        self._pool = pool
        self._running = False
        self._server_config = kwargs

    @property
    def is_running(self) -> bool:
        """Whether the server is currently running."""
        return self._running

    @property
    def pool(self) -> AgentPool[Any]:
        """The underlying agent pool."""
        return self._pool

    async def run(self) -> None:
        """Run the server.

        Template method that handles common lifecycle management
        and delegates to subclass-specific _run() implementation.

        Raises:
            RuntimeError: If server is already running
        """
        if self._running:
            msg = "Server is already running"
            raise RuntimeError(msg)

        logger.info("Starting %s server", self.__class__.__name__)
        self._running = True

        try:
            await self._run()
        except Exception:
            logger.exception("Server error")
            raise
        finally:
            self._running = False
            logger.info("Server %s stopped", self.__class__.__name__)

    @abc.abstractmethod
    async def _run(self) -> None:
        """Run the server implementation.

        Subclasses must implement this method to handle their
        specific protocol setup and serving logic.
        """
        ...

    async def shutdown(self) -> None:
        """Shutdown the server.

        Default implementation just sets running state to False.
        Subclasses can override for custom cleanup logic.
        """
        if self._running:
            logger.info("Shutting down %s server", self.__class__.__name__)
            self._running = False

    async def __aenter__(self) -> Self:
        """Async context manager entry.

        Default implementation does nothing. Subclasses can override
        to perform initialization before the server starts.

        Returns:
            Self for fluent interface
        """
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        """Async context manager exit.

        Default implementation calls shutdown() if server is running.
        Subclasses can override for custom cleanup logic.
        """
        if self._running:
            await self.shutdown()

    def __repr__(self) -> str:
        """String representation of the server."""
        status = "running" if self._running else "stopped"
        pool_info = f"pool-{len(self._pool.agents)}-agents"
        return f"{self.__class__.__name__}({pool_info}, {status})"

is_running property

is_running: bool

Whether the server is currently running.

pool property

pool: AgentPool[Any]

The underlying agent pool.

__aenter__ async

__aenter__() -> Self

Async context manager entry.

Default implementation does nothing. Subclasses can override to perform initialization before the server starts.

Returns:

Type	Description
Self	Self for fluent interface

Source code in src/llmling_agent/server.py

async def __aenter__(self) -> Self:
    """Async context manager entry.

    Default implementation does nothing. Subclasses can override
    to perform initialization before the server starts.

    Returns:
        Self for fluent interface
    """
    return self

__aexit__ async

__aexit__(exc_type, exc_val, exc_tb) -> None

Async context manager exit.

Default implementation calls shutdown() if server is running. Subclasses can override for custom cleanup logic.

Source code in src/llmling_agent/server.py

async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
    """Async context manager exit.

    Default implementation calls shutdown() if server is running.
    Subclasses can override for custom cleanup logic.
    """
    if self._running:
        await self.shutdown()

__init__

__init__(pool: AgentPool[Any], **kwargs: Any)

Initialize the server bridge.

Parameters:

Name	Type	Description	Default
pool	AgentPool[Any]	Agent pool to expose via the protocol	required
**kwargs	Any	Additional configuration options	{}

Source code in src/llmling_agent/server.py

def __init__(
    self,
    pool: AgentPool[Any],
    **kwargs: Any,
):
    """Initialize the server bridge.

    Args:
        pool: Agent pool to expose via the protocol
        **kwargs: Additional configuration options
    """
    self._pool = pool
    self._running = False
    self._server_config = kwargs

__repr__

__repr__() -> str

String representation of the server.

Source code in src/llmling_agent/server.py

def __repr__(self) -> str:
    """String representation of the server."""
    status = "running" if self._running else "stopped"
    pool_info = f"pool-{len(self._pool.agents)}-agents"
    return f"{self.__class__.__name__}({pool_info}, {status})"

run async

run() -> None

Run the server.

Template method that handles common lifecycle management and delegates to subclass-specific _run() implementation.

Raises:

Type	Description
RuntimeError	If server is already running

Source code in src/llmling_agent/server.py

async def run(self) -> None:
    """Run the server.

    Template method that handles common lifecycle management
    and delegates to subclass-specific _run() implementation.

    Raises:
        RuntimeError: If server is already running
    """
    if self._running:
        msg = "Server is already running"
        raise RuntimeError(msg)

    logger.info("Starting %s server", self.__class__.__name__)
    self._running = True

    try:
        await self._run()
    except Exception:
        logger.exception("Server error")
        raise
    finally:
        self._running = False
        logger.info("Server %s stopped", self.__class__.__name__)

shutdown async

shutdown() -> None

Shutdown the server.

Default implementation just sets running state to False. Subclasses can override for custom cleanup logic.

Source code in src/llmling_agent/server.py

async def shutdown(self) -> None:
    """Shutdown the server.

    Default implementation just sets running state to False.
    Subclasses can override for custom cleanup logic.
    """
    if self._running:
        logger.info("Shutting down %s server", self.__class__.__name__)
        self._running = False