nodes

Class info

Classes

Name	Children	Inherits
EventNodeConfig llmling_agent.models.nodes Base configuration for event nodes.		NodeConfig
InputProvider llmling_agent_input.base Base class for handling all UI interactions.		ABC
MCPServerBase llmling_agent.models.mcp_server Base model for MCP server configuration.	StdioMCPServer SSEMCPServer	BaseModel
NodeConfig llmling_agent.models.nodes Configuration for a Node of the messaging system.	EventNodeConfig AgentConfig TeamConfig	BaseModel
StdioMCPServer llmling_agent.models.mcp_server MCP server started via stdio.		MCPServerBase

🛈 DocStrings

Team configuration models.

EventNodeConfig

Bases: NodeConfig

Base configuration for event nodes.

All event node configurations must: 1. Specify their type for discrimination 2. Implement get_event() to create their event instance

Source code in src/llmling_agent/models/nodes.py

class EventNodeConfig(NodeConfig):
    """Base configuration for event nodes.

    All event node configurations must:
    1. Specify their type for discrimination
    2. Implement get_event() to create their event instance
    """

    type: str = Field("event", init=False)
    """Discriminator field for event configs."""

    enabled: bool = True
    """Whether this event source is active."""

    template: str = DEFAULT_TEMPLATE
    """Jinja2 template for formatting events."""

    include_metadata: bool = True
    """Control metadata visibility in template."""

    include_timestamp: bool = True
    """Control timestamp visibility in template."""

    @abstractmethod
    def get_event(self) -> Event[Any]:
        """Create event instance from this configuration.

        This method should:
        1. Create the event instance
        2. Wrap any functions if needed
        3. Return the configured event
        """
        raise NotImplementedError

    def get_event_node(self) -> EventNode[Any]:
        """Create event node from config."""
        from llmling_agent.messaging.eventnode import EventNode

        event = self.get_event()
        return EventNode(
            event=event,
            name=self.name,
            description=self.description,
            mcp_servers=self.mcp_servers,
        )

enabled class-attribute instance-attribute

enabled: bool = True

Whether this event source is active.

include_metadata class-attribute instance-attribute

include_metadata: bool = True

Control metadata visibility in template.

include_timestamp class-attribute instance-attribute

include_timestamp: bool = True

Control timestamp visibility in template.

template class-attribute instance-attribute

template: str = DEFAULT_TEMPLATE

Jinja2 template for formatting events.

type class-attribute instance-attribute

type: str = Field('event', init=False)

Discriminator field for event configs.

get_event abstractmethod

get_event() -> Event[Any]

Create event instance from this configuration.

This method should: 1. Create the event instance 2. Wrap any functions if needed 3. Return the configured event

Source code in src/llmling_agent/models/nodes.py

@abstractmethod
def get_event(self) -> Event[Any]:
    """Create event instance from this configuration.

    This method should:
    1. Create the event instance
    2. Wrap any functions if needed
    3. Return the configured event
    """
    raise NotImplementedError

get_event_node

get_event_node() -> EventNode[Any]

Create event node from config.

Source code in src/llmling_agent/models/nodes.py

def get_event_node(self) -> EventNode[Any]:
    """Create event node from config."""
    from llmling_agent.messaging.eventnode import EventNode

    event = self.get_event()
    return EventNode(
        event=event,
        name=self.name,
        description=self.description,
        mcp_servers=self.mcp_servers,
    )

NodeConfig

Bases: BaseModel

Configuration for a Node of the messaging system.

Source code in src/llmling_agent/models/nodes.py

class NodeConfig(BaseModel):
    """Configuration for a Node of the messaging system."""

    name: str | None = None
    """Name of the Agent / Team"""

    description: str | None = None
    """Optional description of the agent / team."""

    triggers: list[EventConfig] = Field(default_factory=list)
    """Event sources that activate this agent / team"""

    connections: list[ForwardingTarget] = Field(default_factory=list)
    """Targets to forward results to."""

    mcp_servers: list[str | MCPServerConfig] = Field(default_factory=list)
    """List of MCP server configurations:
    - str entries are converted to StdioMCPServer
    - MCPServerConfig for full server configuration
    """

    input_provider: ImportString[InputProvider] | None = None
    """Provider for human-input-handling."""

    # Future extensions:
    # tools: list[str] | None = None
    # """Tools available to all team members."""

    # knowledge: Knowledge | None = None
    # """Knowledge sources shared by all team members."""

    model_config = ConfigDict(
        frozen=True,
        arbitrary_types_allowed=True,
        extra="forbid",
        use_attribute_docstrings=True,
    )

    def get_mcp_servers(self) -> list[MCPServerConfig]:
        """Get processed MCP server configurations.

        Converts string entries to StdioMCPServer configs by splitting
        into command and arguments.

        Returns:
            List of MCPServerConfig instances

        Raises:
            ValueError: If string entry is empty
        """
        configs: list[MCPServerConfig] = []

        for server in self.mcp_servers:
            match server:
                case str():
                    parts = server.split()
                    if not parts:
                        msg = "Empty MCP server command"
                        raise ValueError(msg)

                    configs.append(StdioMCPServer(command=parts[0], args=parts[1:]))
                case MCPServerBase():
                    configs.append(server)

        return configs

connections class-attribute instance-attribute

connections: list[ForwardingTarget] = Field(default_factory=list)

Targets to forward results to.

description class-attribute instance-attribute

description: str | None = None

Optional description of the agent / team.

input_provider class-attribute instance-attribute

input_provider: ImportString[InputProvider] | None = None

Provider for human-input-handling.

mcp_servers class-attribute instance-attribute

mcp_servers: list[str | MCPServerConfig] = Field(default_factory=list)

List of MCP server configurations: - str entries are converted to StdioMCPServer - MCPServerConfig for full server configuration

name class-attribute instance-attribute

name: str | None = None

Name of the Agent / Team

triggers class-attribute instance-attribute

triggers: list[EventConfig] = Field(default_factory=list)

Event sources that activate this agent / team

get_mcp_servers

get_mcp_servers() -> list[MCPServerConfig]

Get processed MCP server configurations.

Converts string entries to StdioMCPServer configs by splitting into command and arguments.

Returns:

Type	Description
list[MCPServerConfig]	List of MCPServerConfig instances

Raises:

Type	Description
ValueError	If string entry is empty

Source code in src/llmling_agent/models/nodes.py

def get_mcp_servers(self) -> list[MCPServerConfig]:
    """Get processed MCP server configurations.

    Converts string entries to StdioMCPServer configs by splitting
    into command and arguments.

    Returns:
        List of MCPServerConfig instances

    Raises:
        ValueError: If string entry is empty
    """
    configs: list[MCPServerConfig] = []

    for server in self.mcp_servers:
        match server:
            case str():
                parts = server.split()
                if not parts:
                    msg = "Empty MCP server command"
                    raise ValueError(msg)

                configs.append(StdioMCPServer(command=parts[0], args=parts[1:]))
            case MCPServerBase():
                configs.append(server)

    return configs