structured

Class info¶

Classes¶

Name	Children	Inherits
`BaseResponseDefinition` llmling_agent_config.result_types Base class for response definitions.	`InlineResponseDefinition` `ImportedResponseDefinition`	Schema
MessageNode llmling_agent.messaging.messagenode Base class for all message processing nodes.	Agent StructuredAgent BaseTeam	MessageEmitter Generic
StructuredAgent llmling_agent.agent.structured Wrapper for Agent that enforces a specific result type.		MessageNode Generic

🛈 DocStrings¶

LLMling integration with PydanticAI for AI-powered resource interaction.

StructuredAgent ¶

Bases: MessageNode[TDeps, TResult]

Wrapper for Agent that enforces a specific result type.

This wrapper ensures the agent always returns results of the specified type. The type can be provided as: - A Python type for validation - A response definition name from the manifest - A complete response definition instance

Source code in src/llmling_agent/agent/structured.py

class StructuredAgent[TDeps, TResult](MessageNode[TDeps, TResult]):
    """Wrapper for Agent that enforces a specific result type.

    This wrapper ensures the agent always returns results of the specified type.
    The type can be provided as:
    - A Python type for validation
    - A response definition name from the manifest
    - A complete response definition instance
    """

    def __init__(
        self,
        agent: Agent[TDeps] | StructuredAgent[TDeps, TResult] | Callable[..., TResult],
        result_type: type[TResult] | str | ResponseDefinition,
        *,
        tool_name: str | None = None,
        tool_description: str | None = None,
    ):
        """Initialize structured agent wrapper.

        Args:
            agent: Base agent to wrap
            result_type: Expected result type:
                - BaseModel / dataclasses
                - Name of response definition in manifest
                - Complete response definition instance
            tool_name: Optional override for tool name
            tool_description: Optional override for tool description

        Raises:
            ValueError: If named response type not found in manifest
        """
        from llmling_agent.agent.agent import Agent

        logger.debug("StructuredAgent.run result_type = %s", result_type)
        match agent:
            case StructuredAgent():
                self._agent: Agent[TDeps] = agent._agent
            case Callable():
                self._agent = Agent[TDeps](provider=agent, name=agent.__name__)
            case Agent():
                self._agent = agent
            case _:
                msg = "Invalid agent type"
                raise ValueError(msg)

        super().__init__(name=self._agent.name)

        self._result_type = to_type(result_type)
        agent.set_result_type(result_type)

        match result_type:
            case type() | str():
                # For types and named definitions, use overrides if provided
                self._agent.set_result_type(
                    result_type,
                    tool_name=tool_name,
                    tool_description=tool_description,
                )
            case BaseResponseDefinition():
                # For response definitions, use as-is
                # (overrides don't apply to complete definitions)
                self._agent.set_result_type(result_type)

    async def __aenter__(self) -> Self:
        """Enter async context and set up MCP servers.

        Called when agent enters its async context. Sets up any configured
        MCP servers and their tools.
        """
        await self._agent.__aenter__()
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ):
        """Exit async context."""
        await self._agent.__aexit__(exc_type, exc_val, exc_tb)

    def __and__(
        self, other: AnyAgent[Any, Any] | Team[Any] | ProcessorCallback[TResult]
    ) -> Team[TDeps]:
        return self._agent.__and__(other)

    def __or__(self, other: Agent | ProcessorCallback | BaseTeam) -> TeamRun:
        return self._agent.__or__(other)

    async def _run(
        self,
        *prompt: AnyPromptType | TResult,
        result_type: type[TResult] | None = None,
        model: ModelType = None,
        tool_choice: str | list[str] | None = None,
        store_history: bool = True,
        message_id: str | None = None,
        conversation_id: str | None = None,
        wait_for_connections: bool | None = None,
    ) -> ChatMessage[TResult]:
        """Run with fixed result type.

        Args:
            prompt: Any prompt-compatible object or structured objects of type TResult
            result_type: Expected result type:
                - BaseModel / dataclasses
                - Name of response definition in manifest
                - Complete response definition instance
            model: Optional model override
            tool_choice: Filter available tools by name
            store_history: Whether the message exchange should be added to the
                           context window
            message_id: Optional message id for the returned message.
                        Automatically generated if not provided.
            conversation_id: Optional conversation id for the returned message.
            wait_for_connections: Whether to wait for all connections to complete
        """
        typ = result_type or self._result_type
        return await self._agent._run(
            *prompt,
            result_type=typ,
            model=model,
            store_history=store_history,
            tool_choice=tool_choice,
            message_id=message_id,
            conversation_id=conversation_id,
            wait_for_connections=wait_for_connections,
        )

    async def validate_against(
        self,
        prompt: str,
        criteria: type[TResult],
        **kwargs: Any,
    ) -> bool:
        """Check if agent's response satisfies stricter criteria."""
        result = await self.run(prompt, **kwargs)
        try:
            criteria.model_validate(result.content.model_dump())  # type: ignore
        except ValidationError:
            return False
        else:
            return True

    def __repr__(self) -> str:
        type_name = getattr(self._result_type, "__name__", str(self._result_type))
        return f"StructuredAgent({self._agent!r}, result_type={type_name})"

    def __prompt__(self) -> str:
        type_name = getattr(self._result_type, "__name__", str(self._result_type))
        base_info = self._agent.__prompt__()
        return f"{base_info}\nStructured output type: {type_name}"

    def __getattr__(self, name: str) -> Any:
        return getattr(self._agent, name)

    @property
    def context(self) -> AgentContext[TDeps]:
        return self._agent.context

    @context.setter
    def context(self, value: Any):
        self._agent.context = value

    @property
    def name(self) -> str:
        return self._agent.name

    @name.setter
    def name(self, value: str):
        self._agent.name = value

    @property
    def tools(self) -> ToolManager:
        return self._agent.tools

    @property
    def conversation(self) -> ConversationManager:
        return self._agent.conversation

    @overload
    def to_structured(
        self,
        result_type: None,
        *,
        tool_name: str | None = None,
        tool_description: str | None = None,
    ) -> Agent[TDeps]: ...

    @overload
    def to_structured[TNewResult](
        self,
        result_type: type[TNewResult] | str | ResponseDefinition,
        *,
        tool_name: str | None = None,
        tool_description: str | None = None,
    ) -> StructuredAgent[TDeps, TNewResult]: ...

    def to_structured[TNewResult](
        self,
        result_type: type[TNewResult] | str | ResponseDefinition | None,
        *,
        tool_name: str | None = None,
        tool_description: str | None = None,
    ) -> Agent[TDeps] | StructuredAgent[TDeps, TNewResult]:
        if result_type is None:
            return self._agent

        return StructuredAgent(
            self._agent,
            result_type=result_type,
            tool_name=tool_name,
            tool_description=tool_description,
        )

    @property
    def stats(self) -> MessageStats:
        return self._agent.stats

    async def run_iter(
        self,
        *prompt_groups: Sequence[AnyPromptType | PIL.Image.Image | os.PathLike[str]],
        **kwargs: Any,
    ) -> AsyncIterator[ChatMessage[Any]]:
        """Forward run_iter to wrapped agent."""
        async for message in self._agent.run_iter(*prompt_groups, **kwargs):
            yield message

    async def run_job(
        self,
        job: Job[TDeps, TResult],
        *,
        store_history: bool = True,
        include_agent_tools: bool = True,
    ) -> ChatMessage[TResult]:
        """Execute a pre-defined job ensuring type compatibility.

        Args:
            job: Job configuration to execute
            store_history: Whether to add job execution to conversation history
            include_agent_tools: Whether to include agent's tools alongside job tools

        Returns:
            Task execution result

        Raises:
            JobError: If job execution fails or types don't match
            ValueError: If job configuration is invalid
        """
        from llmling_agent.tasks import JobError

        # Validate dependency requirement
        if job.required_dependency is not None:  # noqa: SIM102
            if not isinstance(self.context.data, job.required_dependency):
                msg = (
                    f"Agent dependencies ({type(self.context.data)}) "
                    f"don't match job requirement ({job.required_dependency})"
                )
                raise JobError(msg)

        # Validate return type requirement
        if job.required_return_type != self._result_type:
            msg = (
                f"Agent result type ({self._result_type}) "
                f"doesn't match job requirement ({job.required_return_type})"
            )
            raise JobError(msg)

        # Load task knowledge if provided
        if job.knowledge:
            # Add knowledge sources to context
            resources: list[Resource | str] = list(job.knowledge.paths) + list(
                job.knowledge.resources
            )
            for source in resources:
                await self.conversation.load_context_source(source)
            for prompt in job.knowledge.prompts:
                await self.conversation.load_context_source(prompt)

        try:
            # Register task tools temporarily
            tools = job.get_tools()

            # Use temporary tools
            with self._agent.tools.temporary_tools(
                tools, exclusive=not include_agent_tools
            ):
                # Execute job using StructuredAgent's run to maintain type safety
                return await self.run(await job.get_prompt(), store_history=store_history)

        except Exception as e:
            msg = f"Task execution failed: {e}"
            logger.exception(msg)
            raise JobError(msg) from e

    @classmethod
    def from_callback(
        cls,
        callback: ProcessorCallback[TResult],
        *,
        name: str | None = None,
        **kwargs: Any,
    ) -> StructuredAgent[None, TResult]:
        """Create a structured agent from a processing callback.

        Args:
            callback: Function to process messages. Can be:
                - sync or async
                - with or without context
                - with explicit return type
            name: Optional name for the agent
            **kwargs: Additional arguments for agent

        Example:
            ```python
            class AnalysisResult(BaseModel):
                sentiment: float
                topics: list[str]

            def analyze(msg: str) -> AnalysisResult:
                return AnalysisResult(sentiment=0.8, topics=["tech"])

            analyzer = StructuredAgent.from_callback(analyze)
            ```
        """
        from llmling_agent.agent.agent import Agent
        from llmling_agent_providers.callback import CallbackProvider

        name = name or callback.__name__ or "processor"
        provider = CallbackProvider(callback, name=name)
        agent = Agent[None](provider=provider, name=name, **kwargs)
        # Get return type from signature for validation
        hints = get_type_hints(callback)
        return_type = hints.get("return")

        # If async, unwrap from Awaitable
        if (
            return_type
            and hasattr(return_type, "__origin__")
            and return_type.__origin__ is Awaitable
        ):
            return_type = return_type.__args__[0]
        return StructuredAgent[None, TResult](agent, return_type or str)  # type: ignore

    def is_busy(self) -> bool:
        """Check if agent is currently processing tasks."""
        return bool(self._pending_tasks or self._background_task)

    def run_sync(self, *args, **kwargs):
        """Run agent synchronously."""
        return self._agent.run_sync(*args, result_type=self._result_type, **kwargs)

aenter `async` ¶

__aenter__() -> Self

Enter async context and set up MCP servers.

Called when agent enters its async context. Sets up any configured MCP servers and their tools.

Source code in src/llmling_agent/agent/structured.py

async def __aenter__(self) -> Self:
    """Enter async context and set up MCP servers.

    Called when agent enters its async context. Sets up any configured
    MCP servers and their tools.
    """
    await self._agent.__aenter__()
    return self

aexit `async` ¶

__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
)

Exit async context.

Source code in src/llmling_agent/agent/structured.py

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
):
    """Exit async context."""
    await self._agent.__aexit__(exc_type, exc_val, exc_tb)

init ¶

__init__(
    agent: Agent[TDeps] | StructuredAgent[TDeps, TResult] | Callable[..., TResult],
    result_type: type[TResult] | str | ResponseDefinition,
    *,
    tool_name: str | None = None,
    tool_description: str | None = None,
)

Initialize structured agent wrapper.

Parameters:

Name	Type	Description	Default
`agent`	`Agent[TDeps] \| StructuredAgent[TDeps, TResult] \| Callable[..., TResult]`	Base agent to wrap	required
`result_type`	`type[TResult] \| str \| ResponseDefinition`	Expected result type: - BaseModel / dataclasses - Name of response definition in manifest - Complete response definition instance	required
`tool_name`	`str \| None`	Optional override for tool name	`None`
`tool_description`	`str \| None`	Optional override for tool description	`None`

Raises:

Type	Description
`ValueError`	If named response type not found in manifest

Source code in src/llmling_agent/agent/structured.py

def __init__(
    self,
    agent: Agent[TDeps] | StructuredAgent[TDeps, TResult] | Callable[..., TResult],
    result_type: type[TResult] | str | ResponseDefinition,
    *,
    tool_name: str | None = None,
    tool_description: str | None = None,
):
    """Initialize structured agent wrapper.

    Args:
        agent: Base agent to wrap
        result_type: Expected result type:
            - BaseModel / dataclasses
            - Name of response definition in manifest
            - Complete response definition instance
        tool_name: Optional override for tool name
        tool_description: Optional override for tool description

    Raises:
        ValueError: If named response type not found in manifest
    """
    from llmling_agent.agent.agent import Agent

    logger.debug("StructuredAgent.run result_type = %s", result_type)
    match agent:
        case StructuredAgent():
            self._agent: Agent[TDeps] = agent._agent
        case Callable():
            self._agent = Agent[TDeps](provider=agent, name=agent.__name__)
        case Agent():
            self._agent = agent
        case _:
            msg = "Invalid agent type"
            raise ValueError(msg)

    super().__init__(name=self._agent.name)

    self._result_type = to_type(result_type)
    agent.set_result_type(result_type)

    match result_type:
        case type() | str():
            # For types and named definitions, use overrides if provided
            self._agent.set_result_type(
                result_type,
                tool_name=tool_name,
                tool_description=tool_description,
            )
        case BaseResponseDefinition():
            # For response definitions, use as-is
            # (overrides don't apply to complete definitions)
            self._agent.set_result_type(result_type)

from_callback `classmethod` ¶

from_callback(
    callback: ProcessorCallback[TResult], *, name: str | None = None, **kwargs: Any
) -> StructuredAgent[None, TResult]

Create a structured agent from a processing callback.

Parameters:

Name	Type	Description	Default
`callback`	`ProcessorCallback[TResult]`	Function to process messages. Can be: - sync or async - with or without context - with explicit return type	required
`name`	`str \| None`	Optional name for the agent	`None`
`**kwargs`	`Any`	Additional arguments for agent	`{}`

Example

class AnalysisResult(BaseModel):
    sentiment: float
    topics: list[str]

def analyze(msg: str) -> AnalysisResult:
    return AnalysisResult(sentiment=0.8, topics=["tech"])

analyzer = StructuredAgent.from_callback(analyze)

Source code in src/llmling_agent/agent/structured.py

@classmethod
def from_callback(
    cls,
    callback: ProcessorCallback[TResult],
    *,
    name: str | None = None,
    **kwargs: Any,
) -> StructuredAgent[None, TResult]:
    """Create a structured agent from a processing callback.

    Args:
        callback: Function to process messages. Can be:
            - sync or async
            - with or without context
            - with explicit return type
        name: Optional name for the agent
        **kwargs: Additional arguments for agent

    Example:
        ```python
        class AnalysisResult(BaseModel):
            sentiment: float
            topics: list[str]

        def analyze(msg: str) -> AnalysisResult:
            return AnalysisResult(sentiment=0.8, topics=["tech"])

        analyzer = StructuredAgent.from_callback(analyze)
        ```
    """
    from llmling_agent.agent.agent import Agent
    from llmling_agent_providers.callback import CallbackProvider

    name = name or callback.__name__ or "processor"
    provider = CallbackProvider(callback, name=name)
    agent = Agent[None](provider=provider, name=name, **kwargs)
    # Get return type from signature for validation
    hints = get_type_hints(callback)
    return_type = hints.get("return")

    # If async, unwrap from Awaitable
    if (
        return_type
        and hasattr(return_type, "__origin__")
        and return_type.__origin__ is Awaitable
    ):
        return_type = return_type.__args__[0]
    return StructuredAgent[None, TResult](agent, return_type or str)  # type: ignore

is_busy ¶

is_busy() -> bool

Check if agent is currently processing tasks.

Source code in src/llmling_agent/agent/structured.py

def is_busy(self) -> bool:
    """Check if agent is currently processing tasks."""
    return bool(self._pending_tasks or self._background_task)

run_iter `async` ¶

run_iter(
    *prompt_groups: Sequence[AnyPromptType | Image | PathLike[str]], **kwargs: Any
) -> AsyncIterator[ChatMessage[Any]]

Forward run_iter to wrapped agent.

Source code in src/llmling_agent/agent/structured.py

async def run_iter(
    self,
    *prompt_groups: Sequence[AnyPromptType | PIL.Image.Image | os.PathLike[str]],
    **kwargs: Any,
) -> AsyncIterator[ChatMessage[Any]]:
    """Forward run_iter to wrapped agent."""
    async for message in self._agent.run_iter(*prompt_groups, **kwargs):
        yield message

run_job `async` ¶

run_job(
    job: Job[TDeps, TResult],
    *,
    store_history: bool = True,
    include_agent_tools: bool = True,
) -> ChatMessage[TResult]

Execute a pre-defined job ensuring type compatibility.

Parameters:

Name	Type	Description	Default
`job`	`Job[TDeps, TResult]`	Job configuration to execute	required
`store_history`	`bool`	Whether to add job execution to conversation history	`True`
`include_agent_tools`	`bool`	Whether to include agent's tools alongside job tools	`True`

Returns:

Type	Description
`ChatMessage[TResult]`	Task execution result

Raises:

Type	Description
`JobError`	If job execution fails or types don't match
`ValueError`	If job configuration is invalid

Source code in src/llmling_agent/agent/structured.py

async def run_job(
    self,
    job: Job[TDeps, TResult],
    *,
    store_history: bool = True,
    include_agent_tools: bool = True,
) -> ChatMessage[TResult]:
    """Execute a pre-defined job ensuring type compatibility.

    Args:
        job: Job configuration to execute
        store_history: Whether to add job execution to conversation history
        include_agent_tools: Whether to include agent's tools alongside job tools

    Returns:
        Task execution result

    Raises:
        JobError: If job execution fails or types don't match
        ValueError: If job configuration is invalid
    """
    from llmling_agent.tasks import JobError

    # Validate dependency requirement
    if job.required_dependency is not None:  # noqa: SIM102
        if not isinstance(self.context.data, job.required_dependency):
            msg = (
                f"Agent dependencies ({type(self.context.data)}) "
                f"don't match job requirement ({job.required_dependency})"
            )
            raise JobError(msg)

    # Validate return type requirement
    if job.required_return_type != self._result_type:
        msg = (
            f"Agent result type ({self._result_type}) "
            f"doesn't match job requirement ({job.required_return_type})"
        )
        raise JobError(msg)

    # Load task knowledge if provided
    if job.knowledge:
        # Add knowledge sources to context
        resources: list[Resource | str] = list(job.knowledge.paths) + list(
            job.knowledge.resources
        )
        for source in resources:
            await self.conversation.load_context_source(source)
        for prompt in job.knowledge.prompts:
            await self.conversation.load_context_source(prompt)

    try:
        # Register task tools temporarily
        tools = job.get_tools()

        # Use temporary tools
        with self._agent.tools.temporary_tools(
            tools, exclusive=not include_agent_tools
        ):
            # Execute job using StructuredAgent's run to maintain type safety
            return await self.run(await job.get_prompt(), store_history=store_history)

    except Exception as e:
        msg = f"Task execution failed: {e}"
        logger.exception(msg)
        raise JobError(msg) from e

run_sync ¶

run_sync(*args, **kwargs)

Run agent synchronously.

Source code in src/llmling_agent/agent/structured.py

def run_sync(self, *args, **kwargs):
    """Run agent synchronously."""
    return self._agent.run_sync(*args, result_type=self._result_type, **kwargs)

validate_against `async` ¶

validate_against(prompt: str, criteria: type[TResult], **kwargs: Any) -> bool

Check if agent's response satisfies stricter criteria.

Source code in src/llmling_agent/agent/structured.py

async def validate_against(
    self,
    prompt: str,
    criteria: type[TResult],
    **kwargs: Any,
) -> bool:
    """Check if agent's response satisfies stricter criteria."""
    result = await self.run(prompt, **kwargs)
    try:
        criteria.model_validate(result.content.model_dump())  # type: ignore
    except ValidationError:
        return False
    else:
        return True

structured

Class info¶

Classes¶

🛈 DocStrings¶

StructuredAgent ¶

__aenter__ async ¶

__aexit__ async ¶

__init__ ¶

from_callback classmethod ¶

is_busy ¶

run_iter async ¶

run_job async ¶

run_sync ¶

validate_against async ¶

aenter `async` ¶

aexit `async` ¶

init ¶

from_callback `classmethod` ¶

run_iter `async` ¶

run_job `async` ¶

validate_against `async` ¶