ant_ai.core.types

InvocationContext pydantic-model

Bases: BaseModel

Request-scoped execution context. Treat as read-only during a request.

Subclass to carry deployment-specific fields through a run -- tools that declare a ctx parameter receive the instance, and the A2A/ACP entry points build it for you via from_metadata:

class MyContext(InvocationContext):
    tenant: str = ""
    tags: list[str] | None = None

    def trace_attributes(self) -> dict[str, Any]:
        return {**super().trace_attributes(), "tags": self.tags}

A2AServer(..., context_class=MyContext)

Show JSON schema:

{
  "description": "Request-scoped execution context. Treat as read-only during a request.\n\nSubclass to carry deployment-specific fields through a run -- tools that\ndeclare a `ctx` parameter receive the instance, and the A2A/ACP entry\npoints build it for you via `from_metadata`:\n\n    class MyContext(InvocationContext):\n        tenant: str = \"\"\n        tags: list[str] | None = None\n\n        def trace_attributes(self) -> dict[str, Any]:\n            return {**super().trace_attributes(), \"tags\": self.tags}\n\n    A2AServer(..., context_class=MyContext)",
  "properties": {
    "session_id": {
      "title": "Session Id",
      "type": "string"
    },
    "user_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "User Id"
    },
    "llm_settings": {
      "anyOf": [
        {
          "additionalProperties": true,
          "type": "object"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Llm Settings"
    },
    "workflow_settings": {
      "anyOf": [
        {
          "additionalProperties": true,
          "type": "object"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Workflow Settings"
    }
  },
  "required": [
    "session_id"
  ],
  "title": "InvocationContext",
  "type": "object"
}

Config:

• frozen: True

Fields:

• session_id (str)
• user_id (str | None)
• llm_settings (dict[str, Any] | None)
• workflow_settings (dict[str, Any] | None)

Source code in src/ant_ai/core/types.py

class InvocationContext(BaseModel):
    """Request-scoped execution context. Treat as read-only during a request.

    Subclass to carry deployment-specific fields through a run -- tools that
    declare a `ctx` parameter receive the instance, and the A2A/ACP entry
    points build it for you via `from_metadata`:

        class MyContext(InvocationContext):
            tenant: str = ""
            tags: list[str] | None = None

            def trace_attributes(self) -> dict[str, Any]:
                return {**super().trace_attributes(), "tags": self.tags}

        A2AServer(..., context_class=MyContext)
    """

    model_config = ConfigDict(frozen=True)

    session_id: str
    user_id: str | None = Field(default=None)
    llm_settings: dict[str, Any] | None = Field(default=None)
    workflow_settings: dict[str, Any] | None = Field(default=None)

    @classmethod
    def from_metadata(
        cls, *, session_id: str, metadata: Mapping[str, Any] | None = None
    ) -> Self:
        """Build a context from request metadata (e.g. an A2A message's `metadata`).

        Fields are matched by name -- a `user_id` key fills `user_id`, and any field
        a subclass declares is filled the same way. Unknown keys are ignored.
        `session_id` always comes from the argument, never from the metadata.
        Override to map a different wire shape onto your fields.
        """
        return cls.model_validate({**(metadata or {}), "session_id": session_id})

    def outbound_metadata(self) -> dict[str, Any]:
        """What this context forwards when the run calls another agent over A2A.

        Only sent to agents whose `A2AConfig.trusted` is on -- whether a callee
        is trusted is decided per agent, not here. The default is
        every set field except `session_id` (it travels as the A2A `context_id`)
        and `llm_settings` / `workflow_settings`, which are the callee's own.
        Override to withhold more; the callee's `from_metadata` rebuilds it.
        """
        return self.model_dump(
            exclude={"session_id", "llm_settings", "workflow_settings"},
            exclude_none=True,
        )

    def trace_attributes(self) -> dict[str, Any]:
        """Fields bound to the run's trace and sent with `workflow.start`.

        Sinks read these by name: the Langfuse sink, for instance, forwards
        `session_id`, `user_id` and `tags`. Override to add your own; keep
        secrets out, since these end up in whatever observability backend is
        configured.
        """
        return {"session_id": self.session_id, "user_id": self.user_id}

from_metadata classmethod

from_metadata(
    *,
    session_id: str,
    metadata: Mapping[str, Any] | None = None,
) -> Self

Build a context from request metadata (e.g. an A2A message's metadata).

Fields are matched by name -- a user_id key fills user_id, and any field a subclass declares is filled the same way. Unknown keys are ignored. session_id always comes from the argument, never from the metadata. Override to map a different wire shape onto your fields.

Source code in src/ant_ai/core/types.py

@classmethod
def from_metadata(
    cls, *, session_id: str, metadata: Mapping[str, Any] | None = None
) -> Self:
    """Build a context from request metadata (e.g. an A2A message's `metadata`).

    Fields are matched by name -- a `user_id` key fills `user_id`, and any field
    a subclass declares is filled the same way. Unknown keys are ignored.
    `session_id` always comes from the argument, never from the metadata.
    Override to map a different wire shape onto your fields.
    """
    return cls.model_validate({**(metadata or {}), "session_id": session_id})

outbound_metadata

outbound_metadata() -> dict[str, Any]

What this context forwards when the run calls another agent over A2A.

Only sent to agents whose A2AConfig.trusted is on -- whether a callee is trusted is decided per agent, not here. The default is every set field except session_id (it travels as the A2A context_id) and llm_settings / workflow_settings, which are the callee's own. Override to withhold more; the callee's from_metadata rebuilds it.

Source code in src/ant_ai/core/types.py

def outbound_metadata(self) -> dict[str, Any]:
    """What this context forwards when the run calls another agent over A2A.

    Only sent to agents whose `A2AConfig.trusted` is on -- whether a callee
    is trusted is decided per agent, not here. The default is
    every set field except `session_id` (it travels as the A2A `context_id`)
    and `llm_settings` / `workflow_settings`, which are the callee's own.
    Override to withhold more; the callee's `from_metadata` rebuilds it.
    """
    return self.model_dump(
        exclude={"session_id", "llm_settings", "workflow_settings"},
        exclude_none=True,
    )

trace_attributes

trace_attributes() -> dict[str, Any]

Fields bound to the run's trace and sent with workflow.start.

Sinks read these by name: the Langfuse sink, for instance, forwards session_id, user_id and tags. Override to add your own; keep secrets out, since these end up in whatever observability backend is configured.

Source code in src/ant_ai/core/types.py

def trace_attributes(self) -> dict[str, Any]:
    """Fields bound to the run's trace and sent with `workflow.start`.

    Sinks read these by name: the Langfuse sink, for instance, forwards
    `session_id`, `user_id` and `tags`. Override to add your own; keep
    secrets out, since these end up in whatever observability backend is
    configured.
    """
    return {"session_id": self.session_id, "user_id": self.user_id}

State pydantic-model

Bases: BaseModel

Shared mutable state passed through agent steps and workflow nodes.

Subclass to add domain-specific fields:

class MyState(State):
    user_id: str = ""

Show JSON schema:

{
  "$defs": {
    "Message": {
      "description": "Generic message used in a conversation",
      "properties": {
        "kind": {
          "const": "message",
          "default": "message",
          "title": "Kind",
          "type": "string"
        },
        "role": {
          "title": "Role",
          "type": "string"
        },
        "content": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Content"
        },
        "metadata": {
          "additionalProperties": true,
          "title": "Metadata",
          "type": "object"
        }
      },
      "required": [
        "role"
      ],
      "title": "Message",
      "type": "object"
    }
  },
  "description": "Shared mutable state passed through agent steps and workflow nodes.\n\nSubclass to add domain-specific fields:\n\n    class MyState(State):\n        user_id: str = \"\"",
  "properties": {
    "messages": {
      "items": {
        "$ref": "#/$defs/Message"
      },
      "title": "Messages",
      "type": "array"
    },
    "artefacts": {
      "items": {},
      "title": "Artefacts",
      "type": "array"
    }
  },
  "title": "State",
  "type": "object"
}

Fields:

• messages (list[Message])
• artefacts (list[Any])

Source code in src/ant_ai/core/types.py

class State(BaseModel):
    """Shared mutable state passed through agent steps and workflow nodes.

    Subclass to add domain-specific fields:

        class MyState(State):
            user_id: str = ""
    """

    messages: list[Message] = Field(default_factory=list)
    artefacts: list[Any] = Field(default_factory=list)
    _compression_context: list[AnyMessage] | None = PrivateAttr(default=None)

    @property
    def last_message(self) -> Message:
        """Returns the last message in the conversation, if any."""
        if not self.messages:
            raise ValueError("No messages in conversation")
        return self.messages[-1]

    def add_message(self, message: Message) -> None:
        self.messages.append(message)

last_message property

last_message: Message

Returns the last message in the conversation, if any.