models

Class info

Classes

Name	Children	Inherits
AgentConfig llmling_agent.models.agents Configuration for a single agent in the system.		NodeConfig
AgentsManifest llmling_agent.models.manifest Complete agent configuration manifest defining all available agents.		ConfigModel
MCPServerBase llmling_agent.models.mcp_server Base model for MCP server configuration.	StdioMCPServer SSEMCPServer	BaseModel
ResourceInfo llmling_agent.models.resources Information about an available resource.		object
SSEMCPServer llmling_agent.models.mcp_server MCP server using Server-Sent Events transport.		MCPServerBase
SessionQuery llmling_agent.models.session Query configuration for session recovery.		BaseModel
StdioMCPServer llmling_agent.models.mcp_server MCP server started via stdio.		MCPServerBase
TeamConfig llmling_agent.models.teams Configuration for a team or chain of message nodes.		NodeConfig
ToolCallInfo llmling_agent.models.tools Information about an executed tool call.		BaseModel

🛈 DocStrings

Core data models for LLMling agent.

AgentConfig

Bases: NodeConfig

Configuration for a single agent in the system.

Defines an agent's complete configuration including its model, environment, capabilities, and behavior settings. Each agent can have its own: - Language model configuration - Environment setup (tools and resources) - Response type definitions - System prompts and default user prompts - Role-based capabilities

The configuration can be loaded from YAML or created programmatically.

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

class AgentConfig(NodeConfig):
    """Configuration for a single agent in the system.

    Defines an agent's complete configuration including its model, environment,
    capabilities, and behavior settings. Each agent can have its own:
    - Language model configuration
    - Environment setup (tools and resources)
    - Response type definitions
    - System prompts and default user prompts
    - Role-based capabilities

    The configuration can be loaded from YAML or created programmatically.
    """

    provider: ProviderConfig | Literal["pydantic_ai", "human", "litellm"] = "pydantic_ai"
    """Provider configuration or shorthand type"""

    inherits: str | None = None
    """Name of agent config to inherit from"""

    model: str | AnyModelConfig | None = None
    """The model to use for this agent. Can be either a simple model name
    string (e.g. 'openai:gpt-4') or a structured model definition."""

    tools: list[ToolConfig | str] = Field(default_factory=list)
    """A list of tools to register with this agent."""

    toolsets: list[ToolsetConfig] = Field(default_factory=list)
    """Toolset configurations for extensible tool collections."""

    environment: str | AgentEnvironment | None = None
    """Environments configuration (path or object)"""

    capabilities: Capabilities = Field(default_factory=Capabilities)
    """Current agent's capabilities."""

    session: str | SessionQuery | MemoryConfig | None = None
    """Session configuration for conversation recovery."""

    result_type: str | ResponseDefinition | None = None
    """Name of the response definition to use"""

    retries: int = 1
    """Number of retries for failed operations (maps to pydantic-ai's retries)"""

    result_tool_name: str = "final_result"
    """Name of the tool used for structured responses"""

    result_tool_description: str | None = None
    """Custom description for the result tool"""

    result_retries: int | None = None
    """Max retries for result validation"""

    end_strategy: EndStrategy = "early"
    """The strategy for handling multiple tool calls when a final result is found"""

    avatar: str | None = None
    """URL or path to agent's avatar image"""

    system_prompts: list[str] = Field(default_factory=list)
    """System prompts for the agent"""

    library_system_prompts: list[str] = Field(default_factory=list)
    """System prompts for the agent from the library"""

    user_prompts: list[str] = Field(default_factory=list)
    """Default user prompts for the agent"""

    # context_sources: list[ContextSource] = Field(default_factory=list)
    # """Initial context sources to load"""

    config_file_path: str | None = None
    """Config file path for resolving environment."""

    knowledge: Knowledge | None = None
    """Knowledge sources for this agent."""

    workers: list[WorkerConfig] = Field(default_factory=list)
    """Worker agents which will be available as tools."""

    requires_tool_confirmation: ToolConfirmationMode = "per_tool"
    """How to handle tool confirmation:
    - "always": Always require confirmation for all tools
    - "never": Never require confirmation (ignore tool settings)
    - "per_tool": Use individual tool settings
    """

    debug: bool = False
    """Enable debug output for this agent."""

    def is_structured(self) -> bool:
        """Check if this config defines a structured agent."""
        return self.result_type is not None

    @model_validator(mode="before")
    @classmethod
    def normalize_workers(cls, data: dict[str, Any]) -> dict[str, Any]:
        """Convert string workers to WorkerConfig."""
        if workers := data.get("workers"):
            data["workers"] = [
                WorkerConfig.from_str(w)
                if isinstance(w, str)
                else w
                if isinstance(w, WorkerConfig)  # Keep existing WorkerConfig
                else WorkerConfig(**w)  # Convert dict to WorkerConfig
                for w in workers
            ]
        return data

    @model_validator(mode="before")
    @classmethod
    def validate_result_type(cls, data: dict[str, Any]) -> dict[str, Any]:
        """Convert result type and apply its settings."""
        result_type = data.get("result_type")
        if isinstance(result_type, dict):
            # Extract response-specific settings
            tool_name = result_type.pop("result_tool_name", None)
            tool_description = result_type.pop("result_tool_description", None)
            retries = result_type.pop("result_retries", None)

            # Convert remaining dict to ResponseDefinition
            if "type" not in result_type:
                result_type["type"] = "inline"
            data["result_type"] = InlineResponseDefinition(**result_type)

            # Apply extracted settings to agent config
            if tool_name:
                data["result_tool_name"] = tool_name
            if tool_description:
                data["result_tool_description"] = tool_description
            if retries is not None:
                data["result_retries"] = retries

        return data

    @model_validator(mode="before")
    @classmethod
    def handle_model_types(cls, data: dict[str, Any]) -> dict[str, Any]:
        """Convert model inputs to appropriate format."""
        model = data.get("model")
        match model:
            case str():
                data["model"] = {"type": "string", "identifier": model}
        return data

    async def get_toolsets(self) -> list[ResourceProvider]:
        """Get all resource providers for this agent."""
        providers: list[ResourceProvider] = []

        # Add providers from toolsets
        for toolset_config in self.toolsets:
            try:
                provider = toolset_config.get_provider()
                providers.append(provider)
            except Exception as e:
                logger.exception(
                    "Failed to create provider for toolset: %r", toolset_config
                )
                msg = f"Failed to create provider for toolset: {e}"
                raise ValueError(msg) from e

        return providers

    def get_tool_provider(self) -> ResourceProvider | None:
        """Get tool provider for this agent."""
        from llmling_agent.tools.base import ToolInfo

        # Create provider for static tools
        if not self.tools:
            return None
        static_tools: list[ToolInfo] = []
        for tool_config in self.tools:
            try:
                match tool_config:
                    case str():
                        if tool_config.startswith("crewai_tools"):
                            obj = import_class(tool_config)()
                            static_tools.append(ToolInfo.from_crewai_tool(obj))
                        elif tool_config.startswith("langchain"):
                            obj = import_class(tool_config)()
                            static_tools.append(ToolInfo.from_langchain_tool(obj))
                        else:
                            tool = ToolInfo.from_callable(tool_config)
                            static_tools.append(tool)
                    case BaseToolConfig():
                        static_tools.append(tool_config.get_tool())
            except Exception:
                logger.exception("Failed to load tool %r", tool_config)
                continue

        return StaticResourceProvider(name="builtin", tools=static_tools)

    def get_session_config(self) -> MemoryConfig:
        """Get resolved memory configuration."""
        match self.session:
            case str() | UUID():
                return MemoryConfig(session=SessionQuery(name=str(self.session)))
            case SessionQuery():
                return MemoryConfig(session=self.session)
            case MemoryConfig():
                return self.session
            case None:
                return MemoryConfig()

    def get_system_prompts(self) -> list[BasePrompt]:
        """Get all system prompts as BasePrompts."""
        prompts: list[BasePrompt] = []
        for prompt in self.system_prompts:
            match prompt:
                case str():
                    # Convert string to StaticPrompt
                    static_prompt = StaticPrompt(
                        name="system",
                        description="System prompt",
                        messages=[PromptMessage(role="system", content=prompt)],
                    )
                    prompts.append(static_prompt)
                case BasePrompt():
                    prompts.append(prompt)
        return prompts

    def get_provider(self) -> AgentProvider:
        """Get resolved provider instance.

        Creates provider instance based on configuration:
        - Full provider config: Use as-is
        - Shorthand type: Create default provider config
        """
        # If string shorthand is used, convert to default provider config
        from llmling_agent.models.providers import (
            CallbackProviderConfig,
            HumanProviderConfig,
            LiteLLMProviderConfig,
            PydanticAIProviderConfig,
        )

        provider_config = self.provider
        if isinstance(provider_config, str):
            match provider_config:
                case "pydantic_ai":
                    provider_config = PydanticAIProviderConfig(model=self.model)
                case "human":
                    provider_config = HumanProviderConfig()
                case "litellm":
                    provider_config = LiteLLMProviderConfig(
                        model=self.model if isinstance(self.model, str) else None
                    )
                case _:
                    try:
                        fn = import_callable(provider_config)
                        provider_config = CallbackProviderConfig(fn=fn)
                    except Exception:  # noqa: BLE001
                        msg = f"Invalid provider type: {provider_config}"
                        raise ValueError(msg)  # noqa: B904

        # Create provider instance from config
        return provider_config.get_provider()

    def render_system_prompts(self, context: dict[str, Any] | None = None) -> list[str]:
        """Render system prompts with context."""
        if not context:
            # Default context
            context = {"name": self.name, "id": 1, "model": self.model}
        return [render_prompt(p, {"agent": context}) for p in self.system_prompts]

    def get_config(self) -> Config:
        """Get configuration for this agent."""
        match self.environment:
            case None:
                # Create minimal config
                caps = LLMCapabilitiesConfig()
                global_settings = GlobalSettings(llm_capabilities=caps)
                return Config(global_settings=global_settings)
            case str() as path:
                # Backward compatibility: treat as file path
                resolved = self._resolve_environment_path(path, self.config_file_path)
                return Config.from_file(resolved)
            case FileEnvironment(uri=uri) as env:
                # Handle FileEnvironment instance
                resolved = env.get_file_path()
                return Config.from_file(resolved)
            case {"type": "file", "uri": uri}:
                # Handle raw dict matching file environment structure
                return Config.from_file(uri)
            case {"type": "inline", "config": config}:
                return config
            case InlineEnvironment() as config:
                return config
            case _:
                msg = f"Invalid environment configuration: {self.environment}"
                raise ValueError(msg)

    def get_environment_path(self) -> str | None:
        """Get environment file path if available."""
        match self.environment:
            case str() as path:
                return self._resolve_environment_path(path, self.config_file_path)
            case {"type": "file", "uri": uri} | FileEnvironment(uri=uri):
                return uri
            case _:
                return None

    @staticmethod
    def _resolve_environment_path(env: str, config_file_path: str | None = None) -> str:
        """Resolve environment path from config store or relative path."""
        from upath import UPath

        try:
            config_store = ConfigStore()
            return config_store.get_config(env)
        except KeyError:
            if config_file_path:
                base_dir = UPath(config_file_path).parent
                return str(base_dir / env)
            return env

avatar class-attribute instance-attribute

avatar: str | None = None

URL or path to agent's avatar image

capabilities class-attribute instance-attribute

capabilities: Capabilities = Field(default_factory=Capabilities)

Current agent's capabilities.

config_file_path class-attribute instance-attribute

config_file_path: str | None = None

Config file path for resolving environment.

debug class-attribute instance-attribute

debug: bool = False

Enable debug output for this agent.

end_strategy class-attribute instance-attribute

end_strategy: EndStrategy = 'early'

The strategy for handling multiple tool calls when a final result is found

environment class-attribute instance-attribute

environment: str | AgentEnvironment | None = None

Environments configuration (path or object)

inherits class-attribute instance-attribute

inherits: str | None = None

Name of agent config to inherit from

knowledge class-attribute instance-attribute

knowledge: Knowledge | None = None

Knowledge sources for this agent.

library_system_prompts class-attribute instance-attribute

library_system_prompts: list[str] = Field(default_factory=list)

System prompts for the agent from the library

model class-attribute instance-attribute

model: str | AnyModelConfig | None = None

The model to use for this agent. Can be either a simple model name string (e.g. 'openai:gpt-4') or a structured model definition.

provider class-attribute instance-attribute

provider: ProviderConfig | Literal['pydantic_ai', 'human', 'litellm'] = 'pydantic_ai'

Provider configuration or shorthand type

requires_tool_confirmation class-attribute instance-attribute

requires_tool_confirmation: ToolConfirmationMode = 'per_tool'

How to handle tool confirmation: - "always": Always require confirmation for all tools - "never": Never require confirmation (ignore tool settings) - "per_tool": Use individual tool settings

result_retries class-attribute instance-attribute

result_retries: int | None = None

Max retries for result validation

result_tool_description class-attribute instance-attribute

result_tool_description: str | None = None

Custom description for the result tool

result_tool_name class-attribute instance-attribute

result_tool_name: str = 'final_result'

Name of the tool used for structured responses

result_type class-attribute instance-attribute

result_type: str | ResponseDefinition | None = None

Name of the response definition to use

retries class-attribute instance-attribute

retries: int = 1

Number of retries for failed operations (maps to pydantic-ai's retries)

session class-attribute instance-attribute

session: str | SessionQuery | MemoryConfig | None = None

Session configuration for conversation recovery.

system_prompts class-attribute instance-attribute

system_prompts: list[str] = Field(default_factory=list)

System prompts for the agent

tools class-attribute instance-attribute

tools: list[ToolConfig | str] = Field(default_factory=list)

A list of tools to register with this agent.

toolsets class-attribute instance-attribute

toolsets: list[ToolsetConfig] = Field(default_factory=list)

Toolset configurations for extensible tool collections.

user_prompts class-attribute instance-attribute

user_prompts: list[str] = Field(default_factory=list)

Default user prompts for the agent

workers class-attribute instance-attribute

workers: list[WorkerConfig] = Field(default_factory=list)

Worker agents which will be available as tools.

_resolve_environment_path staticmethod

_resolve_environment_path(env: str, config_file_path: str | None = None) -> str

Resolve environment path from config store or relative path.

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

@staticmethod
def _resolve_environment_path(env: str, config_file_path: str | None = None) -> str:
    """Resolve environment path from config store or relative path."""
    from upath import UPath

    try:
        config_store = ConfigStore()
        return config_store.get_config(env)
    except KeyError:
        if config_file_path:
            base_dir = UPath(config_file_path).parent
            return str(base_dir / env)
        return env

get_config

get_config() -> Config

Get configuration for this agent.

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

def get_config(self) -> Config:
    """Get configuration for this agent."""
    match self.environment:
        case None:
            # Create minimal config
            caps = LLMCapabilitiesConfig()
            global_settings = GlobalSettings(llm_capabilities=caps)
            return Config(global_settings=global_settings)
        case str() as path:
            # Backward compatibility: treat as file path
            resolved = self._resolve_environment_path(path, self.config_file_path)
            return Config.from_file(resolved)
        case FileEnvironment(uri=uri) as env:
            # Handle FileEnvironment instance
            resolved = env.get_file_path()
            return Config.from_file(resolved)
        case {"type": "file", "uri": uri}:
            # Handle raw dict matching file environment structure
            return Config.from_file(uri)
        case {"type": "inline", "config": config}:
            return config
        case InlineEnvironment() as config:
            return config
        case _:
            msg = f"Invalid environment configuration: {self.environment}"
            raise ValueError(msg)

get_environment_path

get_environment_path() -> str | None

Get environment file path if available.

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

def get_environment_path(self) -> str | None:
    """Get environment file path if available."""
    match self.environment:
        case str() as path:
            return self._resolve_environment_path(path, self.config_file_path)
        case {"type": "file", "uri": uri} | FileEnvironment(uri=uri):
            return uri
        case _:
            return None

get_provider

get_provider() -> AgentProvider

Get resolved provider instance.

Creates provider instance based on configuration: - Full provider config: Use as-is - Shorthand type: Create default provider config

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

def get_provider(self) -> AgentProvider:
    """Get resolved provider instance.

    Creates provider instance based on configuration:
    - Full provider config: Use as-is
    - Shorthand type: Create default provider config
    """
    # If string shorthand is used, convert to default provider config
    from llmling_agent.models.providers import (
        CallbackProviderConfig,
        HumanProviderConfig,
        LiteLLMProviderConfig,
        PydanticAIProviderConfig,
    )

    provider_config = self.provider
    if isinstance(provider_config, str):
        match provider_config:
            case "pydantic_ai":
                provider_config = PydanticAIProviderConfig(model=self.model)
            case "human":
                provider_config = HumanProviderConfig()
            case "litellm":
                provider_config = LiteLLMProviderConfig(
                    model=self.model if isinstance(self.model, str) else None
                )
            case _:
                try:
                    fn = import_callable(provider_config)
                    provider_config = CallbackProviderConfig(fn=fn)
                except Exception:  # noqa: BLE001
                    msg = f"Invalid provider type: {provider_config}"
                    raise ValueError(msg)  # noqa: B904

    # Create provider instance from config
    return provider_config.get_provider()

get_session_config

get_session_config() -> MemoryConfig

Get resolved memory configuration.

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

def get_session_config(self) -> MemoryConfig:
    """Get resolved memory configuration."""
    match self.session:
        case str() | UUID():
            return MemoryConfig(session=SessionQuery(name=str(self.session)))
        case SessionQuery():
            return MemoryConfig(session=self.session)
        case MemoryConfig():
            return self.session
        case None:
            return MemoryConfig()

get_system_prompts

get_system_prompts() -> list[BasePrompt]

Get all system prompts as BasePrompts.

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

def get_system_prompts(self) -> list[BasePrompt]:
    """Get all system prompts as BasePrompts."""
    prompts: list[BasePrompt] = []
    for prompt in self.system_prompts:
        match prompt:
            case str():
                # Convert string to StaticPrompt
                static_prompt = StaticPrompt(
                    name="system",
                    description="System prompt",
                    messages=[PromptMessage(role="system", content=prompt)],
                )
                prompts.append(static_prompt)
            case BasePrompt():
                prompts.append(prompt)
    return prompts

get_tool_provider

get_tool_provider() -> ResourceProvider | None

Get tool provider for this agent.

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

def get_tool_provider(self) -> ResourceProvider | None:
    """Get tool provider for this agent."""
    from llmling_agent.tools.base import ToolInfo

    # Create provider for static tools
    if not self.tools:
        return None
    static_tools: list[ToolInfo] = []
    for tool_config in self.tools:
        try:
            match tool_config:
                case str():
                    if tool_config.startswith("crewai_tools"):
                        obj = import_class(tool_config)()
                        static_tools.append(ToolInfo.from_crewai_tool(obj))
                    elif tool_config.startswith("langchain"):
                        obj = import_class(tool_config)()
                        static_tools.append(ToolInfo.from_langchain_tool(obj))
                    else:
                        tool = ToolInfo.from_callable(tool_config)
                        static_tools.append(tool)
                case BaseToolConfig():
                    static_tools.append(tool_config.get_tool())
        except Exception:
            logger.exception("Failed to load tool %r", tool_config)
            continue

    return StaticResourceProvider(name="builtin", tools=static_tools)

get_toolsets async

get_toolsets() -> list[ResourceProvider]

Get all resource providers for this agent.

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

async def get_toolsets(self) -> list[ResourceProvider]:
    """Get all resource providers for this agent."""
    providers: list[ResourceProvider] = []

    # Add providers from toolsets
    for toolset_config in self.toolsets:
        try:
            provider = toolset_config.get_provider()
            providers.append(provider)
        except Exception as e:
            logger.exception(
                "Failed to create provider for toolset: %r", toolset_config
            )
            msg = f"Failed to create provider for toolset: {e}"
            raise ValueError(msg) from e

    return providers

handle_model_types classmethod

handle_model_types(data: dict[str, Any]) -> dict[str, Any]

Convert model inputs to appropriate format.

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

@model_validator(mode="before")
@classmethod
def handle_model_types(cls, data: dict[str, Any]) -> dict[str, Any]:
    """Convert model inputs to appropriate format."""
    model = data.get("model")
    match model:
        case str():
            data["model"] = {"type": "string", "identifier": model}
    return data

is_structured

is_structured() -> bool

Check if this config defines a structured agent.

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

def is_structured(self) -> bool:
    """Check if this config defines a structured agent."""
    return self.result_type is not None

normalize_workers classmethod

normalize_workers(data: dict[str, Any]) -> dict[str, Any]

Convert string workers to WorkerConfig.

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

@model_validator(mode="before")
@classmethod
def normalize_workers(cls, data: dict[str, Any]) -> dict[str, Any]:
    """Convert string workers to WorkerConfig."""
    if workers := data.get("workers"):
        data["workers"] = [
            WorkerConfig.from_str(w)
            if isinstance(w, str)
            else w
            if isinstance(w, WorkerConfig)  # Keep existing WorkerConfig
            else WorkerConfig(**w)  # Convert dict to WorkerConfig
            for w in workers
        ]
    return data

render_system_prompts

render_system_prompts(context: dict[str, Any] | None = None) -> list[str]

Render system prompts with context.

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

def render_system_prompts(self, context: dict[str, Any] | None = None) -> list[str]:
    """Render system prompts with context."""
    if not context:
        # Default context
        context = {"name": self.name, "id": 1, "model": self.model}
    return [render_prompt(p, {"agent": context}) for p in self.system_prompts]

validate_result_type classmethod

validate_result_type(data: dict[str, Any]) -> dict[str, Any]

Convert result type and apply its settings.

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

@model_validator(mode="before")
@classmethod
def validate_result_type(cls, data: dict[str, Any]) -> dict[str, Any]:
    """Convert result type and apply its settings."""
    result_type = data.get("result_type")
    if isinstance(result_type, dict):
        # Extract response-specific settings
        tool_name = result_type.pop("result_tool_name", None)
        tool_description = result_type.pop("result_tool_description", None)
        retries = result_type.pop("result_retries", None)

        # Convert remaining dict to ResponseDefinition
        if "type" not in result_type:
            result_type["type"] = "inline"
        data["result_type"] = InlineResponseDefinition(**result_type)

        # Apply extracted settings to agent config
        if tool_name:
            data["result_tool_name"] = tool_name
        if tool_description:
            data["result_tool_description"] = tool_description
        if retries is not None:
            data["result_retries"] = retries

    return data

AgentsManifest

Bases: ConfigModel

Complete agent configuration manifest defining all available agents.

This is the root configuration that: - Defines available response types (both inline and imported) - Configures all agent instances and their settings - Sets up custom role definitions and capabilities - Manages environment configurations

A single manifest can define multiple agents that can work independently or collaborate through the orchestrator.

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

class AgentsManifest(ConfigModel):
    """Complete agent configuration manifest defining all available agents.

    This is the root configuration that:
    - Defines available response types (both inline and imported)
    - Configures all agent instances and their settings
    - Sets up custom role definitions and capabilities
    - Manages environment configurations

    A single manifest can define multiple agents that can work independently
    or collaborate through the orchestrator.
    """

    INHERIT: str | list[str] | None = None
    """Inheritance references."""

    agents: dict[str, AgentConfig] = Field(default_factory=dict)
    """Mapping of agent IDs to their configurations"""

    teams: dict[str, TeamConfig] = Field(default_factory=dict)
    """Mapping of team IDs to their configurations"""

    storage: StorageConfig = Field(default_factory=StorageConfig)
    """Storage provider configuration."""

    observability: ObservabilityConfig = Field(default_factory=ObservabilityConfig)
    """Observability provider configuration."""

    conversion: ConversionConfig = Field(default_factory=ConversionConfig)
    """Document conversion configuration."""

    responses: dict[str, ResponseDefinition] = Field(default_factory=dict)
    """Mapping of response names to their definitions"""

    jobs: dict[str, Job] = Field(default_factory=dict)
    """Pre-defined jobs, ready to be used by nodes."""

    mcp_servers: list[str | MCPServerConfig] = Field(default_factory=list)
    """List of MCP server configurations:

    These MCP servers are used to provide tools and other resources to the nodes.
    """
    pool_server: PoolServerConfig = Field(default_factory=PoolServerConfig)
    """Pool server configuration.

    This MCP server configuration is used for the pool MCP server,
    which exposes pool functionality to other applications / clients."""

    prompts: PromptConfig = Field(default_factory=PromptConfig)

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

    def clone_agent_config(
        self,
        name: str,
        new_name: str | None = None,
        *,
        template_context: dict[str, Any] | None = None,
        **overrides: Any,
    ) -> str:
        """Create a copy of an agent configuration.

        Args:
            name: Name of agent to clone
            new_name: Optional new name (auto-generated if None)
            template_context: Variables for template rendering
            **overrides: Configuration overrides for the clone

        Returns:
            Name of the new agent

        Raises:
            KeyError: If original agent not found
            ValueError: If new name already exists or if overrides invalid
        """
        if name not in self.agents:
            msg = f"Agent {name} not found"
            raise KeyError(msg)

        actual_name = new_name or f"{name}_copy_{len(self.agents)}"
        if actual_name in self.agents:
            msg = f"Agent {actual_name} already exists"
            raise ValueError(msg)

        # Deep copy the configuration
        config = self.agents[name].model_copy(deep=True)

        # Apply overrides
        for key, value in overrides.items():
            if not hasattr(config, key):
                msg = f"Invalid override: {key}"
                raise ValueError(msg)
            setattr(config, key, value)

        # Handle template rendering if context provided
        if template_context:
            # Apply name from context if not explicitly overridden
            if "name" in template_context and "name" not in overrides:
                config.name = template_context["name"]

            # Render system prompts
            config.system_prompts = config.render_system_prompts(template_context)

        self.agents[actual_name] = config
        return actual_name

    @model_validator(mode="before")
    @classmethod
    def resolve_inheritance(cls, data: dict) -> dict:
        """Resolve agent inheritance chains."""
        nodes = data.get("agents", {})
        resolved: dict[str, dict] = {}
        seen: set[str] = set()

        def resolve_node(name: str) -> dict:
            if name in resolved:
                return resolved[name]

            if name in seen:
                msg = f"Circular inheritance detected: {name}"
                raise ValueError(msg)

            seen.add(name)
            config = (
                nodes[name].model_copy()
                if hasattr(nodes[name], "model_copy")
                else nodes[name].copy()
            )
            inherit = (
                config.get("inherits") if isinstance(config, dict) else config.inherits
            )
            if inherit:
                if inherit not in nodes:
                    msg = f"Parent agent {inherit} not found"
                    raise ValueError(msg)

                # Get resolved parent config
                parent = resolve_node(inherit)
                # Merge parent with child (child overrides parent)
                merged = parent.copy()
                merged.update(config)
                config = merged

            seen.remove(name)
            resolved[name] = config
            return config

        # Resolve all nodes
        for name in nodes:
            resolved[name] = resolve_node(name)

        # Update nodes with resolved configs
        data["agents"] = resolved
        return data

    @model_validator(mode="after")
    def set_instrument_libraries(self) -> Self:
        """Auto-set libraries to instrument based on used providers."""
        if (
            not self.observability.enabled
            or self.observability.instrument_libraries is not None
        ):
            return self
        self.observability.instrument_libraries = list(self.get_used_providers())
        return self

    @property
    def node_names(self) -> list[str]:
        """Get list of all agent and team names."""
        return list(self.agents.keys()) + list(self.teams.keys())

    @property
    def nodes(self) -> dict[str, Any]:
        """Get all agent and team configurations."""
        return {**self.agents, **self.teams}

    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

    @cached_property
    def prompt_manager(self) -> PromptManager:
        """Get prompt manager for this manifest."""
        from llmling_agent.prompts.manager import PromptManager

        return PromptManager(self.prompts)

    # @model_validator(mode="after")
    # def validate_response_types(self) -> AgentsManifest:
    #     """Ensure all agent result_types exist in responses or are inline."""
    #     for agent_id, agent in self.agents.items():
    #         if (
    #             isinstance(agent.result_type, str)
    #             and agent.result_type not in self.responses
    #         ):
    #             msg = f"'{agent.result_type=}' for '{agent_id=}' not found in responses"
    #             raise ValueError(msg)
    #     return self

    def get_agent[TAgentDeps](
        self, name: str, deps: TAgentDeps | None = None
    ) -> AnyAgent[TAgentDeps, Any]:
        from llmling import RuntimeConfig

        from llmling_agent import Agent, AgentContext

        config = self.agents[name]
        # Create runtime without async context
        cfg = config.get_config()
        runtime = RuntimeConfig.from_config(cfg)

        # Create context with config path and capabilities
        context = AgentContext[TAgentDeps](
            node_name=name,
            data=deps,
            capabilities=config.capabilities,
            definition=self,
            config=config,
            runtime=runtime,
            # pool=self,
            # confirmation_callback=confirmation_callback,
        )

        sys_prompts = config.system_prompts.copy()
        # Library prompts
        if config.library_system_prompts:
            for prompt_ref in config.library_system_prompts:
                try:
                    content = self.prompt_manager.get_sync(prompt_ref)
                    sys_prompts.append(content)
                except Exception as e:
                    msg = f"Failed to load library prompt {prompt_ref!r} for agent {name}"
                    logger.exception(msg)
                    raise ValueError(msg) from e
        # Create agent with runtime and context
        agent = Agent[Any](
            runtime=runtime,
            context=context,
            provider=config.get_provider(),
            system_prompt=sys_prompts,
            name=name,
            description=config.description,
            retries=config.retries,
            session=config.get_session_config(),
            result_retries=config.result_retries,
            end_strategy=config.end_strategy,
            capabilities=config.capabilities,
            debug=config.debug,
            # name=config.name or name,
        )
        if result_type := self.get_result_type(name):
            return agent.to_structured(result_type)
        return agent

    def get_used_providers(self) -> set[str]:
        """Get all providers configured in this manifest."""
        providers = set[str]()

        for agent_config in self.agents.values():
            match agent_config.provider:
                case "pydantic_ai":
                    providers.add("pydantic_ai")
                case "litellm":
                    providers.add("litellm")
                case BaseProviderConfig():
                    providers.add(agent_config.provider.type)
        return providers

    @classmethod
    def from_file(cls, path: StrPath) -> Self:
        """Load agent configuration from YAML file.

        Args:
            path: Path to the configuration file

        Returns:
            Loaded agent definition

        Raises:
            ValueError: If loading fails
        """
        import yamling

        try:
            data = yamling.load_yaml_file(path, resolve_inherit=True)
            agent_def = cls.model_validate(data)
            # Update all agents with the config file path and ensure names
            agents = {
                name: config.model_copy(update={"config_file_path": str(path)})
                for name, config in agent_def.agents.items()
            }
            return agent_def.model_copy(update={"agents": agents})
        except Exception as exc:
            msg = f"Failed to load agent config from {path}"
            raise ValueError(msg) from exc

    @cached_property
    def pool(self) -> AgentPool:
        """Create an agent pool from this manifest.

        Returns:
            Configured agent pool
        """
        from llmling_agent.delegation import AgentPool

        return AgentPool(manifest=self)

    def get_result_type(self, agent_name: str) -> type[Any] | None:
        """Get the resolved result type for an agent.

        Returns None if no result type is configured.
        """
        agent_config = self.agents[agent_name]
        if not agent_config.result_type:
            return None
        logger.debug("Building response model for %r", agent_config.result_type)
        if isinstance(agent_config.result_type, str):
            response_def = self.responses[agent_config.result_type]
            return response_def.create_model()  # type: ignore
        return agent_config.result_type.create_model()  # type: ignore

INHERIT class-attribute instance-attribute

INHERIT: str | list[str] | None = None

Inheritance references.

agents class-attribute instance-attribute

agents: dict[str, AgentConfig] = Field(default_factory=dict)

Mapping of agent IDs to their configurations

conversion class-attribute instance-attribute

conversion: ConversionConfig = Field(default_factory=ConversionConfig)

Document conversion configuration.

jobs class-attribute instance-attribute

jobs: dict[str, Job] = Field(default_factory=dict)

Pre-defined jobs, ready to be used by nodes.

mcp_servers class-attribute instance-attribute

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

List of MCP server configurations:

These MCP servers are used to provide tools and other resources to the nodes.

node_names property

node_names: list[str]

Get list of all agent and team names.

nodes property

nodes: dict[str, Any]

Get all agent and team configurations.

observability class-attribute instance-attribute

observability: ObservabilityConfig = Field(default_factory=ObservabilityConfig)

Observability provider configuration.

pool cached property

pool: AgentPool

Create an agent pool from this manifest.

Returns:

Type	Description
AgentPool	Configured agent pool

pool_server class-attribute instance-attribute

pool_server: PoolServerConfig = Field(default_factory=PoolServerConfig)

Pool server configuration.

This MCP server configuration is used for the pool MCP server, which exposes pool functionality to other applications / clients.

prompt_manager cached property

prompt_manager: PromptManager

Get prompt manager for this manifest.

responses class-attribute instance-attribute

responses: dict[str, ResponseDefinition] = Field(default_factory=dict)

Mapping of response names to their definitions

storage class-attribute instance-attribute

storage: StorageConfig = Field(default_factory=StorageConfig)

Storage provider configuration.

teams class-attribute instance-attribute

teams: dict[str, TeamConfig] = Field(default_factory=dict)

Mapping of team IDs to their configurations

clone_agent_config

clone_agent_config(
    name: str,
    new_name: str | None = None,
    *,
    template_context: dict[str, Any] | None = None,
    **overrides: Any,
) -> str

Create a copy of an agent configuration.

Parameters:

Name	Type	Description	Default
name	str	Name of agent to clone	required
new_name	str | None	Optional new name (auto-generated if None)	None
template_context	dict[str, Any] | None	Variables for template rendering	None
**overrides	Any	Configuration overrides for the clone	{}

Returns:

Type	Description
str	Name of the new agent

Raises:

Type	Description
KeyError	If original agent not found
ValueError	If new name already exists or if overrides invalid

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

def clone_agent_config(
    self,
    name: str,
    new_name: str | None = None,
    *,
    template_context: dict[str, Any] | None = None,
    **overrides: Any,
) -> str:
    """Create a copy of an agent configuration.

    Args:
        name: Name of agent to clone
        new_name: Optional new name (auto-generated if None)
        template_context: Variables for template rendering
        **overrides: Configuration overrides for the clone

    Returns:
        Name of the new agent

    Raises:
        KeyError: If original agent not found
        ValueError: If new name already exists or if overrides invalid
    """
    if name not in self.agents:
        msg = f"Agent {name} not found"
        raise KeyError(msg)

    actual_name = new_name or f"{name}_copy_{len(self.agents)}"
    if actual_name in self.agents:
        msg = f"Agent {actual_name} already exists"
        raise ValueError(msg)

    # Deep copy the configuration
    config = self.agents[name].model_copy(deep=True)

    # Apply overrides
    for key, value in overrides.items():
        if not hasattr(config, key):
            msg = f"Invalid override: {key}"
            raise ValueError(msg)
        setattr(config, key, value)

    # Handle template rendering if context provided
    if template_context:
        # Apply name from context if not explicitly overridden
        if "name" in template_context and "name" not in overrides:
            config.name = template_context["name"]

        # Render system prompts
        config.system_prompts = config.render_system_prompts(template_context)

    self.agents[actual_name] = config
    return actual_name

from_file classmethod

from_file(path: StrPath) -> Self

Load agent configuration from YAML file.

Parameters:

Name	Type	Description	Default
path	StrPath	Path to the configuration file	required

Returns:

Type	Description
Self	Loaded agent definition

Raises:

Type	Description
ValueError	If loading fails

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

@classmethod
def from_file(cls, path: StrPath) -> Self:
    """Load agent configuration from YAML file.

    Args:
        path: Path to the configuration file

    Returns:
        Loaded agent definition

    Raises:
        ValueError: If loading fails
    """
    import yamling

    try:
        data = yamling.load_yaml_file(path, resolve_inherit=True)
        agent_def = cls.model_validate(data)
        # Update all agents with the config file path and ensure names
        agents = {
            name: config.model_copy(update={"config_file_path": str(path)})
            for name, config in agent_def.agents.items()
        }
        return agent_def.model_copy(update={"agents": agents})
    except Exception as exc:
        msg = f"Failed to load agent config from {path}"
        raise ValueError(msg) from exc

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/manifest.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

get_result_type

get_result_type(agent_name: str) -> type[Any] | None

Get the resolved result type for an agent.

Returns None if no result type is configured.

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

def get_result_type(self, agent_name: str) -> type[Any] | None:
    """Get the resolved result type for an agent.

    Returns None if no result type is configured.
    """
    agent_config = self.agents[agent_name]
    if not agent_config.result_type:
        return None
    logger.debug("Building response model for %r", agent_config.result_type)
    if isinstance(agent_config.result_type, str):
        response_def = self.responses[agent_config.result_type]
        return response_def.create_model()  # type: ignore
    return agent_config.result_type.create_model()  # type: ignore

get_used_providers

get_used_providers() -> set[str]

Get all providers configured in this manifest.

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

def get_used_providers(self) -> set[str]:
    """Get all providers configured in this manifest."""
    providers = set[str]()

    for agent_config in self.agents.values():
        match agent_config.provider:
            case "pydantic_ai":
                providers.add("pydantic_ai")
            case "litellm":
                providers.add("litellm")
            case BaseProviderConfig():
                providers.add(agent_config.provider.type)
    return providers

resolve_inheritance classmethod

resolve_inheritance(data: dict) -> dict

Resolve agent inheritance chains.

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

@model_validator(mode="before")
@classmethod
def resolve_inheritance(cls, data: dict) -> dict:
    """Resolve agent inheritance chains."""
    nodes = data.get("agents", {})
    resolved: dict[str, dict] = {}
    seen: set[str] = set()

    def resolve_node(name: str) -> dict:
        if name in resolved:
            return resolved[name]

        if name in seen:
            msg = f"Circular inheritance detected: {name}"
            raise ValueError(msg)

        seen.add(name)
        config = (
            nodes[name].model_copy()
            if hasattr(nodes[name], "model_copy")
            else nodes[name].copy()
        )
        inherit = (
            config.get("inherits") if isinstance(config, dict) else config.inherits
        )
        if inherit:
            if inherit not in nodes:
                msg = f"Parent agent {inherit} not found"
                raise ValueError(msg)

            # Get resolved parent config
            parent = resolve_node(inherit)
            # Merge parent with child (child overrides parent)
            merged = parent.copy()
            merged.update(config)
            config = merged

        seen.remove(name)
        resolved[name] = config
        return config

    # Resolve all nodes
    for name in nodes:
        resolved[name] = resolve_node(name)

    # Update nodes with resolved configs
    data["agents"] = resolved
    return data

set_instrument_libraries

set_instrument_libraries() -> Self

Auto-set libraries to instrument based on used providers.

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

@model_validator(mode="after")
def set_instrument_libraries(self) -> Self:
    """Auto-set libraries to instrument based on used providers."""
    if (
        not self.observability.enabled
        or self.observability.instrument_libraries is not None
    ):
        return self
    self.observability.instrument_libraries = list(self.get_used_providers())
    return self

MCPServerBase

Bases: BaseModel

Base model for MCP server configuration.

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

class MCPServerBase(BaseModel):
    """Base model for MCP server configuration."""

    type: str
    """Type discriminator for MCP server configurations."""

    name: str | None = None
    """Optional name for referencing the server."""

    enabled: bool = True
    """Whether this server is currently enabled."""

    environment: dict[str, str] | None = None
    """Environment variables to pass to the server process."""

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

    def get_env_vars(self) -> dict[str, str]:
        """Get environment variables for the server process."""
        env = os.environ.copy()
        if self.environment:
            env.update(self.environment)
        env["PYTHONIOENCODING"] = "utf-8"
        return env

enabled class-attribute instance-attribute

enabled: bool = True

Whether this server is currently enabled.

environment class-attribute instance-attribute

environment: dict[str, str] | None = None

Environment variables to pass to the server process.

name class-attribute instance-attribute

name: str | None = None

Optional name for referencing the server.

type instance-attribute

type: str

Type discriminator for MCP server configurations.

get_env_vars

get_env_vars() -> dict[str, str]

Get environment variables for the server process.

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

def get_env_vars(self) -> dict[str, str]:
    """Get environment variables for the server process."""
    env = os.environ.copy()
    if self.environment:
        env.update(self.environment)
    env["PYTHONIOENCODING"] = "utf-8"
    return env

ResourceInfo dataclass

Information about an available resource.

This class provides essential information about a resource that can be loaded. Use the resource name with load_resource() to access the actual content.

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

@dataclass
class ResourceInfo:
    """Information about an available resource.

    This class provides essential information about a resource that can be loaded.
    Use the resource name with load_resource() to access the actual content.
    """

    name: str
    """Name of the resource, use this with load_resource()"""

    uri: str
    """URI identifying the resource location"""

    description: str | None = None
    """Optional description of the resource's content or purpose"""

description class-attribute instance-attribute

description: str | None = None

Optional description of the resource's content or purpose

name instance-attribute

name: str

Name of the resource, use this with load_resource()

uri instance-attribute

uri: str

URI identifying the resource location

SSEMCPServer

Bases: MCPServerBase

MCP server using Server-Sent Events transport.

Connects to a server over HTTP with SSE for real-time communication.

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

class SSEMCPServer(MCPServerBase):
    """MCP server using Server-Sent Events transport.

    Connects to a server over HTTP with SSE for real-time communication.
    """

    type: Literal["sse"] = Field("sse", init=False)
    """SSE server configuration."""

    url: str
    """URL of the SSE server endpoint."""

type class-attribute instance-attribute

type: Literal['sse'] = Field('sse', init=False)

SSE server configuration.

url instance-attribute

url: str

URL of the SSE server endpoint.

SessionQuery

Bases: BaseModel

Query configuration for session recovery.

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

class SessionQuery(BaseModel):
    """Query configuration for session recovery."""

    name: str | None = None
    """Session identifier to match."""

    agents: set[str] | None = None
    """Filter by agent names."""

    since: str | None = None
    """Time period to look back (e.g. "1h", "2d")."""

    until: str | None = None
    """Time period to look up to."""

    contains: str | None = None
    """Filter by message content."""

    roles: set[MessageRole] | None = None
    """Only include specific message roles."""

    limit: int | None = None
    """Maximum number of messages to return."""

    include_forwarded: bool = True
    """Whether to include messages forwarded through agents."""

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

    def get_time_cutoff(self) -> datetime | None:
        """Get datetime from time period string."""
        if not self.since:
            return None
        delta = parse_time_period(self.since)
        return datetime.now() - delta

agents class-attribute instance-attribute

agents: set[str] | None = None

Filter by agent names.

contains class-attribute instance-attribute

contains: str | None = None

Filter by message content.

include_forwarded class-attribute instance-attribute

include_forwarded: bool = True

Whether to include messages forwarded through agents.

limit class-attribute instance-attribute

limit: int | None = None

Maximum number of messages to return.

name class-attribute instance-attribute

name: str | None = None

Session identifier to match.

roles class-attribute instance-attribute

roles: set[MessageRole] | None = None

Only include specific message roles.

since class-attribute instance-attribute

since: str | None = None

Time period to look back (e.g. "1h", "2d").

until class-attribute instance-attribute

until: str | None = None

Time period to look up to.

get_time_cutoff

get_time_cutoff() -> datetime | None

Get datetime from time period string.

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

def get_time_cutoff(self) -> datetime | None:
    """Get datetime from time period string."""
    if not self.since:
        return None
    delta = parse_time_period(self.since)
    return datetime.now() - delta

StdioMCPServer

Bases: MCPServerBase

MCP server started via stdio.

Uses subprocess communication through standard input/output streams.

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

class StdioMCPServer(MCPServerBase):
    """MCP server started via stdio.

    Uses subprocess communication through standard input/output streams.
    """

    type: Literal["stdio"] = Field("stdio", init=False)
    """Stdio server coniguration."""

    command: str
    """Command to execute (e.g. "pipx", "python", "node")."""

    args: list[str] = Field(default_factory=list)
    """Command arguments (e.g. ["run", "some-server", "--debug"])."""

    @classmethod
    def from_string(cls, command: str) -> StdioMCPServer:
        """Create a MCP server from a command string."""
        cmd, args = command.split(maxsplit=1)
        return cls(command=cmd, args=args.split())

args class-attribute instance-attribute

args: list[str] = Field(default_factory=list)

Command arguments (e.g. ["run", "some-server", "--debug"]).

command instance-attribute

command: str

Command to execute (e.g. "pipx", "python", "node").

type class-attribute instance-attribute

type: Literal['stdio'] = Field('stdio', init=False)

Stdio server coniguration.

from_string classmethod

from_string(command: str) -> StdioMCPServer

Create a MCP server from a command string.

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

@classmethod
def from_string(cls, command: str) -> StdioMCPServer:
    """Create a MCP server from a command string."""
    cmd, args = command.split(maxsplit=1)
    return cls(command=cmd, args=args.split())

TeamConfig

Bases: NodeConfig

Configuration for a team or chain of message nodes.

Teams can be either parallel execution groups or sequential chains. They can contain both agents and other teams as members.

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

class TeamConfig(NodeConfig):
    """Configuration for a team or chain of message nodes.

    Teams can be either parallel execution groups or sequential chains.
    They can contain both agents and other teams as members.
    """

    mode: Literal["parallel", "sequential"]
    """Execution mode for team members."""

    members: list[str]
    """Names of agents or other teams that are part of this team."""

    shared_prompt: str | None = None
    """Optional shared prompt for this team."""

members instance-attribute

members: list[str]

Names of agents or other teams that are part of this team.

mode instance-attribute

mode: Literal['parallel', 'sequential']

Execution mode for team members.

shared_prompt class-attribute instance-attribute

shared_prompt: str | None = None

Optional shared prompt for this team.

ToolCallInfo

Bases: BaseModel

Information about an executed tool call.

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

class ToolCallInfo(BaseModel):
    """Information about an executed tool call."""

    tool_name: str
    """Name of the tool that was called."""

    args: dict[str, Any]
    """Arguments passed to the tool."""

    result: Any
    """Result returned by the tool."""

    agent_name: str
    """Name of the calling agent."""

    tool_call_id: str = Field(default_factory=lambda: str(uuid4()))
    """ID provided by the model (e.g. OpenAI function call ID)."""

    timestamp: datetime = Field(default_factory=datetime.now)
    """When the tool was called."""

    message_id: str | None = None
    """ID of the message that triggered this tool call."""

    context_data: Any | None = None
    """Optional context data that was passed to the agent's run() method."""

    error: str | None = None
    """Error message if the tool call failed."""

    timing: float | None = None
    """Time taken for this specific tool call in seconds."""

    agent_tool_name: str | None = None
    """If this tool is agent-based, the name of that agent."""

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

agent_name instance-attribute

agent_name: str

Name of the calling agent.

agent_tool_name class-attribute instance-attribute

agent_tool_name: str | None = None

If this tool is agent-based, the name of that agent.

args instance-attribute

args: dict[str, Any]

Arguments passed to the tool.

context_data class-attribute instance-attribute

context_data: Any | None = None

Optional context data that was passed to the agent's run() method.

error class-attribute instance-attribute

error: str | None = None

Error message if the tool call failed.

message_id class-attribute instance-attribute

message_id: str | None = None

ID of the message that triggered this tool call.

result instance-attribute

result: Any

Result returned by the tool.

timestamp class-attribute instance-attribute

timestamp: datetime = Field(default_factory=now)

When the tool was called.

timing class-attribute instance-attribute

timing: float | None = None

Time taken for this specific tool call in seconds.

tool_call_id class-attribute instance-attribute

tool_call_id: str = Field(default_factory=lambda: str(uuid4()))

ID provided by the model (e.g. OpenAI function call ID).

tool_name instance-attribute

tool_name: str

Name of the tool that was called.