Intent, Response & Tasks¶

Intent¶

Intent `dataclass` ¶

Bases: Generic[TParams]

Parsed intent representing a user's request.

Immutable data class serving as the starting point for agent processing. Generic on a Pydantic payload type TParams so handlers can declare intent: Intent[OrderFilters] to receive a validated, schema- constrained payload as intent.params.

Backward compatibility

Handlers using the bare Intent annotation (or no annotation at all) keep working exactly as before. params defaults to None for legacy handlers, and the legacy parameters: dict[str, Any] field is still populated by both keyword and LLM parsing paths so existing handlers that read intent.parameters['x'] continue to work.

Attributes:

Name	Type	Description
`raw`	`str`	The original natural language request.
`action`	`IntentAction`	The classified action type.
`domain`	`str`	The domain area (e.g., "order", "product", "user").
`params`	`TParams \| None`	Optional typed payload, populated when the handler declares `Intent[T]` and the framework constrained the LLM output to `T`. `None` for legacy handlers.
`parameters`	`dict[str, Any]`	Extracted parameters from the request as a plain dict. Always populated for backward compatibility — when `params` is set, `parameters` mirrors `params.model_dump()`.
`confidence`	`float`	Parsing confidence score (0.0-1.0).
`ambiguities`	`list[str]`	List of detected ambiguities needing clarification.
`session_context`	`dict[str, Any]`	Accumulated session context from prior turns.

Source code in src/agenticapi/interface/intent.py

@dataclass(frozen=True, slots=True)
class Intent(Generic[TParams]):  # noqa: UP046 — explicit Generic preserves Python 3.13 introspection
    """Parsed intent representing a user's request.

    Immutable data class serving as the starting point for agent processing.
    Generic on a Pydantic payload type ``TParams`` so handlers can declare
    ``intent: Intent[OrderFilters]`` to receive a validated, schema-
    constrained payload as ``intent.params``.

    Backward compatibility:
        Handlers using the bare ``Intent`` annotation (or no annotation
        at all) keep working exactly as before. ``params`` defaults to
        ``None`` for legacy handlers, and the legacy
        ``parameters: dict[str, Any]`` field is still populated by both
        keyword and LLM parsing paths so existing handlers that read
        ``intent.parameters['x']`` continue to work.

    Attributes:
        raw: The original natural language request.
        action: The classified action type.
        domain: The domain area (e.g., "order", "product", "user").
        params: Optional typed payload, populated when the handler
            declares ``Intent[T]`` and the framework constrained the
            LLM output to ``T``. ``None`` for legacy handlers.
        parameters: Extracted parameters from the request as a plain
            dict. Always populated for backward compatibility — when
            ``params`` is set, ``parameters`` mirrors
            ``params.model_dump()``.
        confidence: Parsing confidence score (0.0-1.0).
        ambiguities: List of detected ambiguities needing clarification.
        session_context: Accumulated session context from prior turns.
    """

    raw: str
    action: IntentAction
    domain: str
    params: TParams | None = None
    parameters: dict[str, Any] = field(default_factory=dict)
    confidence: float = 1.0
    ambiguities: list[str] = field(default_factory=list)
    session_context: dict[str, Any] = field(default_factory=dict)

    @property
    def is_write(self) -> bool:
        """Whether this intent involves a write or execute operation."""
        return self.action in (IntentAction.WRITE, IntentAction.EXECUTE)

    @property
    def needs_clarification(self) -> bool:
        """Whether this intent has ambiguities requiring user clarification."""
        return self.action == IntentAction.CLARIFY or len(self.ambiguities) > 0

is_write `property` ¶

is_write: bool

Whether this intent involves a write or execute operation.

needs_clarification `property` ¶

needs_clarification: bool

Whether this intent has ambiguities requiring user clarification.

IntentAction¶

IntentAction ¶

Bases: StrEnum

Action type classification for intents.

Source code in src/agenticapi/interface/intent.py

class IntentAction(StrEnum):
    """Action type classification for intents."""

    READ = "read"
    WRITE = "write"
    ANALYZE = "analyze"
    EXECUTE = "execute"
    CLARIFY = "clarify"

IntentParser¶

IntentParser ¶

Parses raw natural language into Intent objects.

Can operate in two modes: - Without LLM: basic keyword-based parsing for action classification and simple domain extraction. - With LLM: uses structured prompts for accurate classification and parameter extraction.

Example

parser = IntentParser() intent = await parser.parse("Show me this month's order count") assert intent.action == IntentAction.READ

parser_llm = IntentParser(llm=backend) intent = await parser_llm.parse("Cancel order #1234") assert intent.action == IntentAction.WRITE

Source code in src/agenticapi/interface/intent.py

class IntentParser:
    """Parses raw natural language into Intent objects.

    Can operate in two modes:
    - Without LLM: basic keyword-based parsing for action classification
      and simple domain extraction.
    - With LLM: uses structured prompts for accurate classification
      and parameter extraction.

    Example:
        parser = IntentParser()
        intent = await parser.parse("Show me this month's order count")
        assert intent.action == IntentAction.READ

        parser_llm = IntentParser(llm=backend)
        intent = await parser_llm.parse("Cancel order #1234")
        assert intent.action == IntentAction.WRITE
    """

    def __init__(self, *, llm: LLMBackend | None = None) -> None:
        """Initialize the intent parser.

        Args:
            llm: Optional LLM backend for advanced parsing. If None,
                falls back to keyword-based parsing.
        """
        self._llm = llm

    async def parse(
        self,
        raw: str,
        *,
        session_context: dict[str, Any] | None = None,
        schema: type[BaseModel] | None = None,
    ) -> Intent[Any]:
        """Parse a natural language request into an Intent.

        Args:
            raw: The raw natural language request string.
            session_context: Optional accumulated session context.
            schema: Optional Pydantic model the LLM should produce a
                payload for. When supplied, the parser asks the LLM
                backend to constrain its output to the model's JSON
                schema (provider-native structured output) and the
                returned :class:`Intent` has ``intent.params`` populated
                with a validated instance of ``schema``. When omitted,
                ``params`` stays ``None`` and only the legacy
                ``parameters`` dict is populated.

        Returns:
            A parsed Intent object. ``Intent.params`` is set when
            ``schema`` was provided.

        Raises:
            IntentParseError: If parsing fails completely or the LLM
                response fails schema validation after one retry.
        """
        if not raw or not raw.strip():
            raise IntentParseError("Empty intent string")

        ctx = session_context or {}

        if self._llm is not None:
            return await self._parse_with_llm(raw, ctx, schema=schema)

        # Keyword-only path: schema parameter is honoured by attempting
        # a best-effort validation of the empty-default model. If the
        # schema has any required fields, the call raises so the
        # caller can fall back to using the bare ``parameters`` dict.
        intent = self._parse_with_keywords(raw, ctx)
        if schema is not None:
            try:
                params = schema.model_validate({})
            except ValidationError:
                # Required fields without an LLM — best we can do is
                # return the legacy intent unchanged.
                return intent
            return Intent(
                raw=intent.raw,
                action=intent.action,
                domain=intent.domain,
                params=params,
                parameters=params.model_dump(),
                confidence=intent.confidence,
                ambiguities=intent.ambiguities,
                session_context=intent.session_context,
            )
        return intent

    async def _parse_with_llm(
        self,
        raw: str,
        session_context: dict[str, Any],
        *,
        schema: type[BaseModel] | None = None,
    ) -> Intent[Any]:
        """Parse intent using the LLM backend.

        Args:
            raw: The raw request string.
            session_context: Session context dict.
            schema: Optional Pydantic model to constrain the LLM output
                via the backend's native structured-output API.

        Returns:
            Parsed Intent.

        Raises:
            IntentParseError: If LLM call or JSON parsing fails.
        """
        import time as _time

        from agenticapi.observability import (
            AgenticAPIAttributes,
            GenAIAttributes,
            SpanNames,
            get_tracer,
            record_llm_usage,
        )

        prompt = build_intent_parsing_prompt(raw)
        assert self._llm is not None  # Guaranteed by caller

        # Attach the typed-payload schema to the prompt so backends
        # can constrain output via their native structured-output API.
        # When schema is None this is a no-op and the prompt path stays
        # identical to the legacy behaviour.
        if schema is not None:
            prompt = _attach_response_schema(prompt, schema)

        tracer = get_tracer()
        with tracer.start_as_current_span(SpanNames.GEN_AI_CHAT.value) as llm_span:
            llm_span.set_attribute(GenAIAttributes.OPERATION_NAME.value, "intent_parse")
            llm_span.set_attribute(GenAIAttributes.REQUEST_MODEL.value, self._llm.model_name)
            llm_span.set_attribute(GenAIAttributes.REQUEST_MAX_TOKENS.value, prompt.max_tokens)
            llm_span.set_attribute(GenAIAttributes.REQUEST_TEMPERATURE.value, prompt.temperature)
            if schema is not None:
                llm_span.set_attribute(AgenticAPIAttributes.INTENT_PAYLOAD_SCHEMA.value, schema.__name__)
            llm_started = _time.monotonic()
            try:
                response = await self._llm.generate(prompt)
            except Exception as exc:
                logger.error("intent_parse_llm_failed", error=str(exc), raw=raw[:200])
                llm_span.record_exception(exc)
                raise IntentParseError(f"LLM call failed during intent parsing: {exc}") from exc

            llm_span.set_attribute(GenAIAttributes.RESPONSE_MODEL.value, response.model)
            llm_span.set_attribute(GenAIAttributes.USAGE_INPUT_TOKENS.value, response.usage.input_tokens)
            llm_span.set_attribute(GenAIAttributes.USAGE_OUTPUT_TOKENS.value, response.usage.output_tokens)
            record_llm_usage(
                model=response.model or self._llm.model_name,
                input_tokens=response.usage.input_tokens,
                output_tokens=response.usage.output_tokens,
                latency_seconds=_time.monotonic() - llm_started,
            )

        # Schema-constrained path: the entire response is the validated
        # payload. Validate, retry once on failure, then either return
        # the typed Intent or fall back to keyword parsing.
        if schema is not None:
            return self._build_typed_intent(
                raw=raw,
                response_content=response.content,
                schema=schema,
                session_context=session_context,
            )

        try:
            parsed = _parse_llm_json(response.content)
        except (json.JSONDecodeError, KeyError, ValueError) as exc:
            logger.warning(
                "intent_parse_json_failed",
                raw_response=response.content[:200],
                error=str(exc),
            )
            # Fall back to keyword parsing if JSON extraction fails
            fallback_intent = self._parse_with_keywords(raw, session_context)
            return Intent(
                raw=fallback_intent.raw,
                action=fallback_intent.action,
                domain=fallback_intent.domain,
                parameters=fallback_intent.parameters,
                confidence=fallback_intent.confidence,
                ambiguities=[*fallback_intent.ambiguities, f"LLM parsing failed: {exc}; used keyword fallback"],
                session_context=fallback_intent.session_context,
            )

        action_str = parsed.get("action", "read")
        try:
            action = IntentAction(action_str)
        except ValueError:
            action = IntentAction.READ

        domain = parsed.get("domain", "general")
        parameters = parsed.get("parameters", {})
        confidence = float(parsed.get("confidence", 0.8))
        ambiguities = parsed.get("ambiguities", [])

        intent: Intent[Any] = Intent(
            raw=raw,
            action=action,
            domain=domain,
            parameters=parameters if isinstance(parameters, dict) else {},
            confidence=max(0.0, min(1.0, confidence)),
            ambiguities=ambiguities if isinstance(ambiguities, list) else [],
            session_context=session_context,
        )

        logger.info(
            "intent_parsed",
            intent_action=intent.action,
            intent_domain=intent.domain,
            confidence=intent.confidence,
            ambiguity_count=len(intent.ambiguities),
        )

        return intent

    def _build_typed_intent(
        self,
        *,
        raw: str,
        response_content: str,
        schema: type[BaseModel],
        session_context: dict[str, Any],
    ) -> Intent[Any]:
        """Validate ``response_content`` against ``schema`` and build a typed Intent.

        Falls back to keyword parsing on validation failure.
        """
        try:
            payload = json.loads(response_content)
        except json.JSONDecodeError as exc:
            logger.warning(
                "intent_typed_parse_json_failed",
                raw_response=response_content[:200],
                error=str(exc),
            )
            return self._typed_keyword_fallback(raw, schema, session_context, str(exc))

        try:
            validated = schema.model_validate(payload)
        except ValidationError as exc:
            logger.warning(
                "intent_typed_validation_failed",
                schema=schema.__name__,
                errors=exc.errors()[:3],
            )
            return self._typed_keyword_fallback(raw, schema, session_context, str(exc))

        # Successful validation — build the typed Intent. We classify
        # the action via the keyword fallback so existing IntentScope
        # rules continue to work even on the typed path.
        keyword_intent = self._parse_with_keywords(raw, session_context)
        intent: Intent[Any] = Intent(
            raw=raw,
            action=keyword_intent.action,
            domain=keyword_intent.domain,
            params=validated,
            parameters=validated.model_dump(),
            confidence=0.95,
            ambiguities=[],
            session_context=session_context,
        )
        logger.info(
            "intent_parsed_typed",
            schema=schema.__name__,
            intent_action=intent.action,
            intent_domain=intent.domain,
        )
        return intent

    def _typed_keyword_fallback(
        self,
        raw: str,
        schema: type[BaseModel],
        session_context: dict[str, Any],
        reason: str,
    ) -> Intent[Any]:
        """Last-ditch typed-intent fallback when LLM output cannot validate.

        Tries to construct ``schema`` from defaults; if that also fails,
        returns the keyword-only intent with no ``params``.
        """
        keyword_intent = self._parse_with_keywords(raw, session_context)
        try:
            params = schema.model_validate({})
            return Intent(
                raw=raw,
                action=keyword_intent.action,
                domain=keyword_intent.domain,
                params=params,
                parameters=params.model_dump(),
                confidence=0.4,
                ambiguities=[f"typed schema fallback: {reason[:120]}"],
                session_context=session_context,
            )
        except ValidationError:
            return Intent(
                raw=keyword_intent.raw,
                action=keyword_intent.action,
                domain=keyword_intent.domain,
                parameters=keyword_intent.parameters,
                confidence=0.3,
                ambiguities=[f"typed schema {schema.__name__} could not be satisfied: {reason[:120]}"],
                session_context=session_context,
            )

    def _parse_with_keywords(
        self,
        raw: str,
        session_context: dict[str, Any],
    ) -> Intent[Any]:
        """Parse intent using simple keyword matching.

        Args:
            raw: The raw request string.
            session_context: Session context dict.

        Returns:
            Parsed Intent with keyword-based classification.
        """
        lower = raw.lower()
        words = set(_WORD_PATTERN.findall(lower))

        action = self._classify_action(words)
        domain = self._extract_domain(words)

        intent: Intent[Any] = Intent(
            raw=raw,
            action=action,
            domain=domain,
            parameters={},
            confidence=0.5,
            ambiguities=[],
            session_context=session_context,
        )

        logger.info(
            "intent_parsed_keywords",
            intent_action=intent.action,
            intent_domain=intent.domain,
            confidence=intent.confidence,
        )

        return intent

    @staticmethod
    def _classify_action(words: set[str]) -> IntentAction:
        """Classify intent action from keyword overlap.

        Args:
            words: Set of lowercase words from the request.

        Returns:
            The best matching IntentAction.
        """
        scores: dict[IntentAction, int] = {
            IntentAction.READ: len(words & _READ_KEYWORDS),
            IntentAction.WRITE: len(words & _WRITE_KEYWORDS),
            IntentAction.ANALYZE: len(words & _ANALYZE_KEYWORDS),
            IntentAction.EXECUTE: len(words & _EXECUTE_KEYWORDS),
        }

        best_action = max(scores, key=lambda k: scores[k])
        if scores[best_action] == 0:
            return IntentAction.READ  # Default

        return best_action

    @staticmethod
    def _extract_domain(words: set[str]) -> str:
        """Extract domain from words using common domain names.

        Args:
            words: Set of lowercase words from the request.

        Returns:
            Extracted domain name or "general".
        """
        known_domains: dict[str, str] = {
            "order": "order",
            "orders": "order",
            "product": "product",
            "products": "product",
            "user": "user",
            "users": "user",
            "customer": "customer",
            "customers": "customer",
            "payment": "payment",
            "payments": "payment",
            "invoice": "invoice",
            "invoices": "invoice",
            "inventory": "inventory",
            "shipping": "shipping",
            "delivery": "shipping",
            "注文": "order",
            "商品": "product",
            "ユーザ": "user",
            "顧客": "customer",
            "支払い": "payment",
        }

        for word in words:
            if word in known_domains:
                return known_domains[word]

        return "general"

init ¶

__init__(*, llm: LLMBackend | None = None) -> None

Initialize the intent parser.

Parameters:

Name	Type	Description	Default
`llm`	`LLMBackend \| None`	Optional LLM backend for advanced parsing. If None, falls back to keyword-based parsing.	`None`

Source code in src/agenticapi/interface/intent.py

def __init__(self, *, llm: LLMBackend | None = None) -> None:
    """Initialize the intent parser.

    Args:
        llm: Optional LLM backend for advanced parsing. If None,
            falls back to keyword-based parsing.
    """
    self._llm = llm

parse `async` ¶

parse(
    raw: str,
    *,
    session_context: dict[str, Any] | None = None,
    schema: type[BaseModel] | None = None,
) -> Intent[Any]

Parse a natural language request into an Intent.

Parameters:

Name	Type	Description	Default
`raw`	`str`	The raw natural language request string.	required
`session_context`	`dict[str, Any] \| None`	Optional accumulated session context.	`None`
`schema`	`type[BaseModel] \| None`	Optional Pydantic model the LLM should produce a payload for. When supplied, the parser asks the LLM backend to constrain its output to the model's JSON schema (provider-native structured output) and the returned :class:`Intent` has `intent.params` populated with a validated instance of `schema`. When omitted, `params` stays `None` and only the legacy `parameters` dict is populated.	`None`

Returns:

Type	Description
`Intent[Any]`	A parsed Intent object. `Intent.params` is set when
`Intent[Any]`	`schema` was provided.

Raises:

Type	Description
`IntentParseError`	If parsing fails completely or the LLM response fails schema validation after one retry.

Source code in src/agenticapi/interface/intent.py

async def parse(
    self,
    raw: str,
    *,
    session_context: dict[str, Any] | None = None,
    schema: type[BaseModel] | None = None,
) -> Intent[Any]:
    """Parse a natural language request into an Intent.

    Args:
        raw: The raw natural language request string.
        session_context: Optional accumulated session context.
        schema: Optional Pydantic model the LLM should produce a
            payload for. When supplied, the parser asks the LLM
            backend to constrain its output to the model's JSON
            schema (provider-native structured output) and the
            returned :class:`Intent` has ``intent.params`` populated
            with a validated instance of ``schema``. When omitted,
            ``params`` stays ``None`` and only the legacy
            ``parameters`` dict is populated.

    Returns:
        A parsed Intent object. ``Intent.params`` is set when
        ``schema`` was provided.

    Raises:
        IntentParseError: If parsing fails completely or the LLM
            response fails schema validation after one retry.
    """
    if not raw or not raw.strip():
        raise IntentParseError("Empty intent string")

    ctx = session_context or {}

    if self._llm is not None:
        return await self._parse_with_llm(raw, ctx, schema=schema)

    # Keyword-only path: schema parameter is honoured by attempting
    # a best-effort validation of the empty-default model. If the
    # schema has any required fields, the call raises so the
    # caller can fall back to using the bare ``parameters`` dict.
    intent = self._parse_with_keywords(raw, ctx)
    if schema is not None:
        try:
            params = schema.model_validate({})
        except ValidationError:
            # Required fields without an LLM — best we can do is
            # return the legacy intent unchanged.
            return intent
        return Intent(
            raw=intent.raw,
            action=intent.action,
            domain=intent.domain,
            params=params,
            parameters=params.model_dump(),
            confidence=intent.confidence,
            ambiguities=intent.ambiguities,
            session_context=intent.session_context,
        )
    return intent

IntentScope¶

IntentScope ¶

Bases: BaseModel

Declarative scope for allowed and denied intents on an endpoint.

Uses wildcard matching (e.g., "order.*" matches "order.create").

Attributes:

Name	Type	Description
`allowed_intents`	`list[str]`	Patterns of allowed intents. ["*"] means all allowed.
`denied_intents`	`list[str]`	Patterns of denied intents. Takes precedence over allowed.

Source code in src/agenticapi/interface/intent.py

class IntentScope(BaseModel):
    """Declarative scope for allowed and denied intents on an endpoint.

    Uses wildcard matching (e.g., "order.*" matches "order.create").

    Attributes:
        allowed_intents: Patterns of allowed intents. ["*"] means all allowed.
        denied_intents: Patterns of denied intents. Takes precedence over allowed.
    """

    model_config = {"extra": "forbid"}

    allowed_intents: list[str] = Field(default_factory=lambda: ["*"])
    denied_intents: list[str] = Field(default_factory=list)

    def matches(self, intent: Intent[Any]) -> bool:
        """Check whether an intent is allowed by this scope.

        Denied patterns take precedence over allowed patterns.
        The intent key used for matching is "{domain}.{action}".

        Args:
            intent: The intent to check.

        Returns:
            True if the intent is allowed by this scope.
        """
        intent_key = f"{intent.domain}.{intent.action}"

        # Check denied first (takes precedence)
        for pattern in self.denied_intents:
            if fnmatch(intent_key, pattern):
                return False

        # Check allowed
        return any(fnmatch(intent_key, pattern) for pattern in self.allowed_intents)

matches ¶

matches(intent: Intent[Any]) -> bool

Check whether an intent is allowed by this scope.

Denied patterns take precedence over allowed patterns. The intent key used for matching is "{domain}.{action}".

Parameters:

Name	Type	Description	Default
`intent`	`Intent[Any]`	The intent to check.	required

Returns:

Type	Description
`bool`	True if the intent is allowed by this scope.

Source code in src/agenticapi/interface/intent.py

def matches(self, intent: Intent[Any]) -> bool:
    """Check whether an intent is allowed by this scope.

    Denied patterns take precedence over allowed patterns.
    The intent key used for matching is "{domain}.{action}".

    Args:
        intent: The intent to check.

    Returns:
        True if the intent is allowed by this scope.
    """
    intent_key = f"{intent.domain}.{intent.action}"

    # Check denied first (takes precedence)
    for pattern in self.denied_intents:
        if fnmatch(intent_key, pattern):
            return False

    # Check allowed
    return any(fnmatch(intent_key, pattern) for pattern in self.allowed_intents)

AgentResponse¶

AgentResponse `dataclass` ¶

Response from an agent endpoint.

Captures the result, status, and metadata of an agent operation.

Attributes:

Name	Type	Description
`result`	`Any`	The primary output from the agent operation.
`status`	`str`	Response status. One of "completed", "pending_approval", "error", or "clarification_needed".
`generated_code`	`str \| None`	The code that was generated and executed (if any).
`reasoning`	`str \| None`	LLM reasoning for the generated code (if any).
`confidence`	`float`	Confidence in the result (0.0-1.0).
`execution_trace_id`	`str \| None`	Identifier for the audit trace of this operation.
`follow_up_suggestions`	`list[str]`	Suggested follow-up actions for the user.
`error`	`str \| None`	Error message if status is "error".
`approval_request`	`dict[str, Any] \| None`	Approval request details if status is "pending_approval".

Source code in src/agenticapi/interface/response.py

@dataclass(slots=True)
class AgentResponse:
    """Response from an agent endpoint.

    Captures the result, status, and metadata of an agent operation.

    Attributes:
        result: The primary output from the agent operation.
        status: Response status. One of "completed", "pending_approval",
            "error", or "clarification_needed".
        generated_code: The code that was generated and executed (if any).
        reasoning: LLM reasoning for the generated code (if any).
        confidence: Confidence in the result (0.0-1.0).
        execution_trace_id: Identifier for the audit trace of this operation.
        follow_up_suggestions: Suggested follow-up actions for the user.
        error: Error message if status is "error".
        approval_request: Approval request details if status is "pending_approval".
    """

    result: Any
    status: str = "completed"
    generated_code: str | None = None
    reasoning: str | None = None
    confidence: float = 1.0
    execution_trace_id: str | None = None
    follow_up_suggestions: list[str] = field(default_factory=list)
    error: str | None = None
    approval_request: dict[str, Any] | None = None

    def to_dict(self) -> dict[str, Any]:
        """Serialize the response to a JSON-compatible dictionary.

        Excludes None values for cleaner output.

        Returns:
            A dictionary representation of the response.
        """
        raw = asdict(self)
        # Remove None values for cleaner JSON output
        return {k: v for k, v in raw.items() if v is not None}

to_dict ¶

to_dict() -> dict[str, Any]

Serialize the response to a JSON-compatible dictionary.

Excludes None values for cleaner output.

Returns:

Type	Description
`dict[str, Any]`	A dictionary representation of the response.

Source code in src/agenticapi/interface/response.py

def to_dict(self) -> dict[str, Any]:
    """Serialize the response to a JSON-compatible dictionary.

    Excludes None values for cleaner output.

    Returns:
        A dictionary representation of the response.
    """
    raw = asdict(self)
    # Remove None values for cleaner JSON output
    return {k: v for k, v in raw.items() if v is not None}

ResponseFormatter¶

ResponseFormatter ¶

Formats AgentResponse for different output formats.

Provides methods for JSON and plain-text serialization.

Example

formatter = ResponseFormatter() json_dict = formatter.format_json(response) text = formatter.format_text(response)

Source code in src/agenticapi/interface/response.py

class ResponseFormatter:
    """Formats AgentResponse for different output formats.

    Provides methods for JSON and plain-text serialization.

    Example:
        formatter = ResponseFormatter()
        json_dict = formatter.format_json(response)
        text = formatter.format_text(response)
    """

    def format_json(self, response: AgentResponse) -> dict[str, Any]:
        """Format the response as a JSON-compatible dictionary.

        Includes all fields, filtering out None values.

        Args:
            response: The agent response to format.

        Returns:
            A JSON-compatible dictionary.
        """
        return response.to_dict()

    def format_text(self, response: AgentResponse) -> str:
        """Format the response as human-readable plain text.

        Args:
            response: The agent response to format.

        Returns:
            A human-readable text representation.
        """
        parts: list[str] = []

        parts.append(f"Status: {response.status}")

        if response.error:
            parts.append(f"Error: {response.error}")
        elif response.result is not None:
            parts.append(f"Result: {response.result}")

        if response.reasoning:
            parts.append(f"Reasoning: {response.reasoning}")

        if response.confidence < 1.0:
            parts.append(f"Confidence: {response.confidence:.2f}")

        if response.follow_up_suggestions:
            parts.append("Suggestions:")
            for suggestion in response.follow_up_suggestions:
                parts.append(f"  - {suggestion}")

        if response.execution_trace_id:
            parts.append(f"Trace ID: {response.execution_trace_id}")

        return "\n".join(parts)

format_json ¶

format_json(response: AgentResponse) -> dict[str, Any]

Format the response as a JSON-compatible dictionary.

Includes all fields, filtering out None values.

Parameters:

Name	Type	Description	Default
`response`	`AgentResponse`	The agent response to format.	required

Returns:

Type	Description
`dict[str, Any]`	A JSON-compatible dictionary.

Source code in src/agenticapi/interface/response.py

def format_json(self, response: AgentResponse) -> dict[str, Any]:
    """Format the response as a JSON-compatible dictionary.

    Includes all fields, filtering out None values.

    Args:
        response: The agent response to format.

    Returns:
        A JSON-compatible dictionary.
    """
    return response.to_dict()

format_text ¶

format_text(response: AgentResponse) -> str

Format the response as human-readable plain text.

Parameters:

Name	Type	Description	Default
`response`	`AgentResponse`	The agent response to format.	required

Returns:

Type	Description
`str`	A human-readable text representation.

Source code in src/agenticapi/interface/response.py

def format_text(self, response: AgentResponse) -> str:
    """Format the response as human-readable plain text.

    Args:
        response: The agent response to format.

    Returns:
        A human-readable text representation.
    """
    parts: list[str] = []

    parts.append(f"Status: {response.status}")

    if response.error:
        parts.append(f"Error: {response.error}")
    elif response.result is not None:
        parts.append(f"Result: {response.result}")

    if response.reasoning:
        parts.append(f"Reasoning: {response.reasoning}")

    if response.confidence < 1.0:
        parts.append(f"Confidence: {response.confidence:.2f}")

    if response.follow_up_suggestions:
        parts.append("Suggestions:")
        for suggestion in response.follow_up_suggestions:
            parts.append(f"  - {suggestion}")

    if response.execution_trace_id:
        parts.append(f"Trace ID: {response.execution_trace_id}")

    return "\n".join(parts)

SessionManager¶

SessionManager ¶

In-memory session manager with TTL-based expiration.

Manages creation, retrieval, update, and deletion of sessions. Sessions are stored in a simple dictionary keyed by session ID.

Example

manager = SessionManager(ttl_seconds=1800) session = await manager.get_or_create(None) # Creates new session.add_turn(intent_raw="hello", response_summary="hi") await manager.update(session)

Source code in src/agenticapi/interface/session.py

class SessionManager:
    """In-memory session manager with TTL-based expiration.

    Manages creation, retrieval, update, and deletion of sessions.
    Sessions are stored in a simple dictionary keyed by session ID.

    Example:
        manager = SessionManager(ttl_seconds=1800)
        session = await manager.get_or_create(None)  # Creates new
        session.add_turn(intent_raw="hello", response_summary="hi")
        await manager.update(session)
    """

    _CLEANUP_INTERVAL = 100  # Run cleanup every N get_or_create calls

    def __init__(self, *, ttl_seconds: int = _DEFAULT_TTL_SECONDS) -> None:
        """Initialize the session manager.

        Args:
            ttl_seconds: Default TTL for new sessions in seconds.
        """
        self._ttl_seconds = ttl_seconds
        self._sessions: dict[str, Session] = {}
        self._access_count = 0

    async def get_or_create(self, session_id: str | None) -> Session:
        """Get an existing session or create a new one.

        If session_id is None, a new session is always created.
        If the session exists but is expired, it is deleted and a
        new one is created with the same ID.

        Args:
            session_id: The session ID to look up, or None for a new session.

        Returns:
            The existing or newly created Session.
        """
        # Periodically clean up expired sessions to prevent memory leaks
        self._access_count += 1
        if self._access_count % self._CLEANUP_INTERVAL == 0:
            self._cleanup_expired()

        if session_id is not None:
            existing = self._sessions.get(session_id)
            if existing is not None:
                if existing.is_expired:
                    logger.info("session_expired", session_id=session_id)
                    del self._sessions[session_id]
                else:
                    existing.last_accessed = datetime.now(tz=UTC)
                    return existing

        # Create new session
        new_id = session_id if session_id is not None else uuid.uuid4().hex
        now = datetime.now(tz=UTC)
        session = Session(
            session_id=new_id,
            created_at=now,
            last_accessed=now,
            ttl_seconds=self._ttl_seconds,
        )
        self._sessions[new_id] = session

        logger.info("session_created", session_id=new_id)
        return session

    async def get(self, session_id: str) -> Session | None:
        """Get an existing session by ID.

        Returns None if the session does not exist or is expired.

        Args:
            session_id: The session ID to look up.

        Returns:
            The Session if found and not expired, otherwise None.
        """
        session = self._sessions.get(session_id)
        if session is None:
            return None
        if session.is_expired:
            del self._sessions[session_id]
            return None
        return session

    async def update(self, session: Session) -> None:
        """Update a session in the store.

        Args:
            session: The session to update.

        Raises:
            SessionError: If the session is not found in the store.
        """
        if session.session_id not in self._sessions:
            raise SessionError(f"Session '{session.session_id}' not found")
        self._sessions[session.session_id] = session

    async def delete(self, session_id: str) -> None:
        """Delete a session from the store.

        Silently succeeds if the session does not exist.

        Args:
            session_id: The session ID to delete.
        """
        self._sessions.pop(session_id, None)
        logger.info("session_deleted", session_id=session_id)

    @property
    def active_count(self) -> int:
        """Number of non-expired sessions currently stored."""
        return sum(1 for s in self._sessions.values() if not s.is_expired)

    def _cleanup_expired(self) -> None:
        """Remove all expired sessions from the store.

        Called periodically from get_or_create() to prevent
        memory leaks from unreferenced expired sessions.
        """
        expired_ids = [sid for sid, s in self._sessions.items() if s.is_expired]
        for sid in expired_ids:
            del self._sessions[sid]
        if expired_ids:
            logger.info("sessions_cleanup", removed_count=len(expired_ids))

active_count `property` ¶

active_count: int

Number of non-expired sessions currently stored.

init ¶

__init__(
    *, ttl_seconds: int = _DEFAULT_TTL_SECONDS
) -> None

Initialize the session manager.

Parameters:

Name	Type	Description	Default
`ttl_seconds`	`int`	Default TTL for new sessions in seconds.	`_DEFAULT_TTL_SECONDS`

Source code in src/agenticapi/interface/session.py

def __init__(self, *, ttl_seconds: int = _DEFAULT_TTL_SECONDS) -> None:
    """Initialize the session manager.

    Args:
        ttl_seconds: Default TTL for new sessions in seconds.
    """
    self._ttl_seconds = ttl_seconds
    self._sessions: dict[str, Session] = {}
    self._access_count = 0

get_or_create `async` ¶

get_or_create(session_id: str | None) -> Session

Get an existing session or create a new one.

If session_id is None, a new session is always created. If the session exists but is expired, it is deleted and a new one is created with the same ID.

Parameters:

Name	Type	Description	Default
`session_id`	`str \| None`	The session ID to look up, or None for a new session.	required

Returns:

Type	Description
`Session`	The existing or newly created Session.

Source code in src/agenticapi/interface/session.py

async def get_or_create(self, session_id: str | None) -> Session:
    """Get an existing session or create a new one.

    If session_id is None, a new session is always created.
    If the session exists but is expired, it is deleted and a
    new one is created with the same ID.

    Args:
        session_id: The session ID to look up, or None for a new session.

    Returns:
        The existing or newly created Session.
    """
    # Periodically clean up expired sessions to prevent memory leaks
    self._access_count += 1
    if self._access_count % self._CLEANUP_INTERVAL == 0:
        self._cleanup_expired()

    if session_id is not None:
        existing = self._sessions.get(session_id)
        if existing is not None:
            if existing.is_expired:
                logger.info("session_expired", session_id=session_id)
                del self._sessions[session_id]
            else:
                existing.last_accessed = datetime.now(tz=UTC)
                return existing

    # Create new session
    new_id = session_id if session_id is not None else uuid.uuid4().hex
    now = datetime.now(tz=UTC)
    session = Session(
        session_id=new_id,
        created_at=now,
        last_accessed=now,
        ttl_seconds=self._ttl_seconds,
    )
    self._sessions[new_id] = session

    logger.info("session_created", session_id=new_id)
    return session

get `async` ¶

get(session_id: str) -> Session | None

Get an existing session by ID.

Returns None if the session does not exist or is expired.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session ID to look up.	required

Returns:

Type	Description
`Session \| None`	The Session if found and not expired, otherwise None.

Source code in src/agenticapi/interface/session.py

async def get(self, session_id: str) -> Session | None:
    """Get an existing session by ID.

    Returns None if the session does not exist or is expired.

    Args:
        session_id: The session ID to look up.

    Returns:
        The Session if found and not expired, otherwise None.
    """
    session = self._sessions.get(session_id)
    if session is None:
        return None
    if session.is_expired:
        del self._sessions[session_id]
        return None
    return session

update `async` ¶

update(session: Session) -> None

Update a session in the store.

Parameters:

Name	Type	Description	Default
`session`	`Session`	The session to update.	required

Raises:

Type	Description
`SessionError`	If the session is not found in the store.

Source code in src/agenticapi/interface/session.py

async def update(self, session: Session) -> None:
    """Update a session in the store.

    Args:
        session: The session to update.

    Raises:
        SessionError: If the session is not found in the store.
    """
    if session.session_id not in self._sessions:
        raise SessionError(f"Session '{session.session_id}' not found")
    self._sessions[session.session_id] = session

delete `async` ¶

delete(session_id: str) -> None

Delete a session from the store.

Silently succeeds if the session does not exist.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session ID to delete.	required

Source code in src/agenticapi/interface/session.py

async def delete(self, session_id: str) -> None:
    """Delete a session from the store.

    Silently succeeds if the session does not exist.

    Args:
        session_id: The session ID to delete.
    """
    self._sessions.pop(session_id, None)
    logger.info("session_deleted", session_id=session_id)

Session¶

Session `dataclass` ¶

A user session for multi-turn agent conversations.

Tracks conversation history and accumulated context across multiple turns. Sessions expire after a configurable TTL.

Attributes:

Name	Type	Description
`session_id`	`str`	Unique identifier for this session.
`created_at`	`datetime`	When the session was created.
`last_accessed`	`datetime`	When the session was last accessed.
`context`	`dict[str, Any]`	Accumulated context dictionary from prior turns.
`history`	`list[dict[str, Any]]`	List of conversation turn records.
`ttl_seconds`	`int`	Time-to-live in seconds before expiration.

Source code in src/agenticapi/interface/session.py

@dataclass(slots=True)
class Session:
    """A user session for multi-turn agent conversations.

    Tracks conversation history and accumulated context across
    multiple turns. Sessions expire after a configurable TTL.

    Attributes:
        session_id: Unique identifier for this session.
        created_at: When the session was created.
        last_accessed: When the session was last accessed.
        context: Accumulated context dictionary from prior turns.
        history: List of conversation turn records.
        ttl_seconds: Time-to-live in seconds before expiration.
    """

    session_id: str
    created_at: datetime
    last_accessed: datetime
    context: dict[str, Any] = field(default_factory=dict)
    history: list[dict[str, Any]] = field(default_factory=list)
    ttl_seconds: int = _DEFAULT_TTL_SECONDS

    def add_turn(self, *, intent_raw: str, response_summary: str) -> None:
        """Record a conversation turn in the session history.

        Also updates the last_accessed timestamp.

        Args:
            intent_raw: The raw user request for this turn.
            response_summary: A summary of the agent's response.
        """
        self.last_accessed = datetime.now(tz=UTC)
        self.history.append(
            {
                "intent": intent_raw,
                "response": response_summary,
                "timestamp": self.last_accessed.isoformat(),
            }
        )

    @property
    def is_expired(self) -> bool:
        """Whether this session has exceeded its TTL."""
        elapsed = datetime.now(tz=UTC) - self.last_accessed
        return elapsed > timedelta(seconds=self.ttl_seconds)

    @property
    def turn_count(self) -> int:
        """Number of conversation turns in this session."""
        return len(self.history)

is_expired `property` ¶

is_expired: bool

Whether this session has exceeded its TTL.

turn_count `property` ¶

turn_count: int

Number of conversation turns in this session.

add_turn ¶

add_turn(*, intent_raw: str, response_summary: str) -> None

Record a conversation turn in the session history.

Also updates the last_accessed timestamp.

Parameters:

Name	Type	Description	Default
`intent_raw`	`str`	The raw user request for this turn.	required
`response_summary`	`str`	A summary of the agent's response.	required

Source code in src/agenticapi/interface/session.py

def add_turn(self, *, intent_raw: str, response_summary: str) -> None:
    """Record a conversation turn in the session history.

    Also updates the last_accessed timestamp.

    Args:
        intent_raw: The raw user request for this turn.
        response_summary: A summary of the agent's response.
    """
    self.last_accessed = datetime.now(tz=UTC)
    self.history.append(
        {
            "intent": intent_raw,
            "response": response_summary,
            "timestamp": self.last_accessed.isoformat(),
        }
    )

AgentTasks¶

AgentTasks ¶

Accumulator for background tasks that run after the response is sent.

Analogous to FastAPI's BackgroundTasks. Agent endpoint handlers receive an AgentTasks instance as a parameter and call add_task() to schedule work that should happen after the HTTP response is returned to the client.

Tasks execute sequentially in the order they were added. If a task fails, subsequent tasks still execute (errors are logged).

Example

async def send_email(to: str, subject: str) -> None: ...

@app.agent_endpoint(name="signup") async def signup(intent: Intent, context: AgentContext, tasks: AgentTasks): tasks.add_task(send_email, to="user@example.com", subject="Welcome!") return {"status": "signed up"}

Source code in src/agenticapi/interface/tasks.py

class AgentTasks:
    """Accumulator for background tasks that run after the response is sent.

    Analogous to FastAPI's ``BackgroundTasks``. Agent endpoint handlers
    receive an ``AgentTasks`` instance as a parameter and call
    ``add_task()`` to schedule work that should happen after the HTTP
    response is returned to the client.

    Tasks execute sequentially in the order they were added. If a task
    fails, subsequent tasks still execute (errors are logged).

    Example:
        async def send_email(to: str, subject: str) -> None:
            ...

        @app.agent_endpoint(name="signup")
        async def signup(intent: Intent, context: AgentContext, tasks: AgentTasks):
            tasks.add_task(send_email, to="user@example.com", subject="Welcome!")
            return {"status": "signed up"}
    """

    def __init__(self) -> None:
        """Initialize an empty task list."""
        self._tasks: list[tuple[Callable[..., Any], tuple[Any, ...], dict[str, Any]]] = []

    def add_task(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> None:
        """Schedule a background task.

        The task will execute after the HTTP response is sent.
        Both sync and async callables are supported.

        Args:
            func: The callable to execute. Can be sync or async.
            *args: Positional arguments for the callable.
            **kwargs: Keyword arguments for the callable.
        """
        self._tasks.append((func, args, kwargs))

    async def execute(self) -> None:
        """Execute all accumulated tasks sequentially.

        Called by the framework after the response is sent. Individual
        task failures are logged but do not prevent subsequent tasks
        from running.
        """
        for func, args, kwargs in self._tasks:
            try:
                result = func(*args, **kwargs)
                if asyncio.iscoroutine(result):
                    await result
            except Exception as exc:
                logger.error(
                    "background_task_failed",
                    task=getattr(func, "__name__", str(func)),
                    error=str(exc),
                )

    @property
    def pending_count(self) -> int:
        """Number of tasks waiting to execute."""
        return len(self._tasks)

pending_count `property` ¶

pending_count: int

Number of tasks waiting to execute.

init ¶

__init__() -> None

Initialize an empty task list.

Source code in src/agenticapi/interface/tasks.py

def __init__(self) -> None:
    """Initialize an empty task list."""
    self._tasks: list[tuple[Callable[..., Any], tuple[Any, ...], dict[str, Any]]] = []

add_task ¶

add_task(
    func: Callable[..., Any], *args: Any, **kwargs: Any
) -> None

Schedule a background task.

The task will execute after the HTTP response is sent. Both sync and async callables are supported.

Parameters:

Name	Type	Description	Default
`func`	`Callable[..., Any]`	The callable to execute. Can be sync or async.	required
`*args`	`Any`	Positional arguments for the callable.	`()`
`**kwargs`	`Any`	Keyword arguments for the callable.	`{}`

Source code in src/agenticapi/interface/tasks.py

def add_task(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> None:
    """Schedule a background task.

    The task will execute after the HTTP response is sent.
    Both sync and async callables are supported.

    Args:
        func: The callable to execute. Can be sync or async.
        *args: Positional arguments for the callable.
        **kwargs: Keyword arguments for the callable.
    """
    self._tasks.append((func, args, kwargs))

execute `async` ¶

execute() -> None

Execute all accumulated tasks sequentially.

Called by the framework after the response is sent. Individual task failures are logged but do not prevent subsequent tasks from running.

Source code in src/agenticapi/interface/tasks.py

async def execute(self) -> None:
    """Execute all accumulated tasks sequentially.

    Called by the framework after the response is sent. Individual
    task failures are logged but do not prevent subsequent tasks
    from running.
    """
    for func, args, kwargs in self._tasks:
        try:
            result = func(*args, **kwargs)
            if asyncio.iscoroutine(result):
                await result
        except Exception as exc:
            logger.error(
                "background_task_failed",
                task=getattr(func, "__name__", str(func)),
                error=str(exc),
            )

Intent, Response & Tasks¶

Intent¶

Intent dataclass ¶

is_write property ¶

needs_clarification property ¶

IntentAction¶

IntentAction ¶

IntentParser¶

IntentParser ¶

__init__ ¶

parse async ¶

IntentScope¶

IntentScope ¶

matches ¶

AgentResponse¶

AgentResponse dataclass ¶

to_dict ¶

ResponseFormatter¶

ResponseFormatter ¶

format_json ¶

format_text ¶

SessionManager¶

SessionManager ¶

active_count property ¶

__init__ ¶

get_or_create async ¶

get async ¶

update async ¶

delete async ¶

Session¶

Session dataclass ¶

is_expired property ¶

turn_count property ¶

add_turn ¶

AgentTasks¶

AgentTasks ¶

pending_count property ¶

__init__ ¶

add_task ¶

execute async ¶

Intent `dataclass` ¶

is_write `property` ¶

needs_clarification `property` ¶

init ¶

parse `async` ¶

AgentResponse `dataclass` ¶

active_count `property` ¶

init ¶

get_or_create `async` ¶

get `async` ¶

update `async` ¶

delete `async` ¶

Session `dataclass` ¶

is_expired `property` ¶

turn_count `property` ¶

pending_count `property` ¶

init ¶

execute `async` ¶