Skip to content

Agent

Optional install

The agent module is not part of the core aitaem install. See Getting Started for install instructions and a two-line quick start for QueryBot and DefinitionBot.

Primitives

Bot

aitaem.agent.base.Bot

Bases: ABC

Abstract base class for all aitaem agent bots.

Subclasses implement _build_agent() to return a configured pydantic-ai Agent. Bots are constructed per user/session; conversation state (message history and result store) lives on the bot instance.

The standard pattern for subclass construction

class MyBot(Bot): def init(self, my_resource, kwargs): self._my_resource = my_resource # set BEFORE super().init() super().init(kwargs) # triggers _build_agent()

def _build_agent(self):
    # self._my_resource is already available here
    from pydantic_ai.toolsets import FunctionToolset

    toolset = FunctionToolset()
    toolset.add_function(my_default_tool)
    for tool in self._tools:
        _register_tool(toolset, tool)
    self._toolset = toolset  # REQUIRED — see contract below

    return pydantic_ai.Agent(..., toolsets=[self._toolset])

Tool composition contract: _build_agent() MUST build a FunctionToolset, register self._tools onto it (via the module-level _register_tool() helper), and assign it to self._toolset before returning the Agent. add_tool() mutates self._toolset in place, so it has nothing to mutate otherwise. Bot.init raises TypeError immediately after _build_agent() returns if self._toolset is still None, naming the offending subclass.

Context-window management: for long-running sessions where history may exceed the model's context limit, pass a history processor via the capabilities argument when constructing the Agent in _build_agent():

from pydantic_ai.capabilities import ProcessHistory, ReinjectSystemPrompt

def _build_agent(self):
    return Agent(
        model=self._model,
        capabilities=[
            ReinjectSystemPrompt(replace_existing=True),
            ProcessHistory(trim_old_messages),
        ],
    )

The processor callable receives the full message list before each model request (including mid-tool-call-loop steps) and returns a modified list. IMPORTANT: only trim complete tool-call pairs (ToolCallPart + matching ToolReturnPart) as a unit. Dropping a ToolReturnPart without its ToolCallPart violates provider API constraints. See pydantic-ai issue #2050. No built-in trimmer is provided by pydantic-ai; implement one in Phase 2+.

Source code in aitaem/agent/base.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
class Bot(ABC):
    """Abstract base class for all aitaem agent bots.

    Subclasses implement _build_agent() to return a configured pydantic-ai
    Agent. Bots are constructed per user/session; conversation state (message
    history and result store) lives on the bot instance.

    The standard pattern for subclass construction:
        class MyBot(Bot):
            def __init__(self, my_resource, **kwargs):
                self._my_resource = my_resource  # set BEFORE super().__init__()
                super().__init__(**kwargs)        # triggers _build_agent()

            def _build_agent(self):
                # self._my_resource is already available here
                from pydantic_ai.toolsets import FunctionToolset

                toolset = FunctionToolset()
                toolset.add_function(my_default_tool)
                for tool in self._tools:
                    _register_tool(toolset, tool)
                self._toolset = toolset  # REQUIRED — see contract below

                return pydantic_ai.Agent(..., toolsets=[self._toolset])

    Tool composition contract: _build_agent() MUST build a FunctionToolset,
    register self._tools onto it (via the module-level _register_tool()
    helper), and assign it to self._toolset before returning the Agent.
    add_tool() mutates self._toolset in place, so it has nothing to mutate
    otherwise. Bot.__init__ raises TypeError immediately after _build_agent()
    returns if self._toolset is still None, naming the offending subclass.

    Context-window management: for long-running sessions where history may
    exceed the model's context limit, pass a history processor via the
    capabilities argument when constructing the Agent in _build_agent():

        from pydantic_ai.capabilities import ProcessHistory, ReinjectSystemPrompt

        def _build_agent(self):
            return Agent(
                model=self._model,
                capabilities=[
                    ReinjectSystemPrompt(replace_existing=True),
                    ProcessHistory(trim_old_messages),
                ],
            )

    The processor callable receives the full message list before each model
    request (including mid-tool-call-loop steps) and returns a modified list.
    IMPORTANT: only trim complete tool-call pairs (ToolCallPart + matching
    ToolReturnPart) as a unit. Dropping a ToolReturnPart without its
    ToolCallPart violates provider API constraints. See pydantic-ai issue #2050.
    No built-in trimmer is provided by pydantic-ai; implement one in Phase 2+.
    """

    def __init__(
        self,
        *,
        model: str,
        tools: list[Any] | None = None,
    ) -> None:
        self._model = model
        self._tools: list[Any] = list(tools or [])
        self._store = ResultStore()
        self._message_history: list[Any] = []
        self._runtime_added_tool_names: list[str] = []
        # Contract: subclass _build_agent() MUST assign a FunctionToolset here before
        # returning — enforced by the check below. If you're reading this in a
        # debugger because self._toolset is None, your _build_agent() didn't set it.
        self._toolset: Any = None
        self._agent = self._build_agent()
        if self._toolset is None:
            raise TypeError(
                f"{type(self).__name__}._build_agent() did not set self._toolset. "
                "Concrete Bot subclasses must build a FunctionToolset, register "
                "self._tools onto it, and assign it to self._toolset before "
                "returning the Agent."
            )

    @abstractmethod
    def _build_agent(self) -> Agent:
        """Return a configured pydantic-ai Agent for this bot."""

    async def chat(
        self,
        message: str,
        *,
        extra_tools: list[Any] | None = None,
    ) -> BotResponse:
        """Send a message and accumulate history (multi-turn entry point)."""
        raise NotImplementedError(
            "chat() must be implemented by the convenience bot subclass."
        )

    async def ask(
        self,
        message: str,
        *,
        extra_tools: list[Any] | None = None,
    ) -> BotResponse:
        """Send a single-turn message without accumulating history."""
        raise NotImplementedError(
            "ask() must be implemented by the convenience bot subclass."
        )

    def add_tool(self, tool: Any) -> None:
        """Add a tool to this bot's persistent tool set at runtime.

        Takes effect on the next chat()/ask() call. Mutations during an
        in-progress run() are undefined.
        """
        before = set(self._toolset.tools)
        _register_tool(self._toolset, tool)
        self._runtime_added_tool_names.extend(sorted(set(self._toolset.tools) - before))

    def add_bot(self, bot: Bot) -> None:
        """Register another bot as a tool (sugar for add_tool(other.as_tool()))."""
        self.add_tool(bot.as_tool())

    def as_tool(self) -> Any:
        """Return a pydantic-ai Tool that wraps this bot's ask() method."""
        raise NotImplementedError("as_tool() implemented in Phase 5.")

    def get_result(self, result_id: str) -> ResultEntry:
        """Retrieve a stored computation result by ID."""
        return self._store.get(result_id)

    def dump_history(self) -> dict[str, Any]:
        """Serialize conversation history and result artifacts to a JSON-safe dict.

        The returned dict is JSON-serializable (suitable for json.dumps). Ibis refs
        are not serialized; only Arrow artifacts are preserved. Reload with
        load_history() to restore history with Arrow artifacts available via
        get_result(), but get_ibis() will return None on restored entries.

        Tools added at runtime via add_tool() are NOT restored by load_history()
        — their names are recorded in the bundle so load_history() can warn if
        they're missing after reload, but the callables themselves aren't
        portably serializable. Pass them again via tools=[...] or call
        add_tool() on the reloaded bot to restore them.
        """
        from aitaem.agent.history import make_bundle

        return make_bundle(
            self._message_history, self._store, self._runtime_added_tool_names
        )

    @classmethod
    def load_history(cls, data: dict[str, Any], **kwargs: Any) -> Bot:
        """Construct a new bot pre-loaded with a serialized history bundle.

        Args:
            data: Bundle returned by dump_history() on a prior bot instance.
            **kwargs: Constructor arguments for the concrete Bot subclass.

        Returns:
            A new bot instance with _message_history and store populated from
            data. The bot's _agent is rebuilt fresh from **kwargs.

        Warns (UserWarning) if the bundle references tools added via
        add_tool() on the original bot that are not present on the reloaded
        bot — pass them again via tools=[...] in **kwargs, or call add_tool()
        after reload, to silence the warning.
        """
        from aitaem.agent.history import load_bundle

        bot = cls(**kwargs)
        bot._message_history = load_bundle(data, bot._store)
        missing = set(data.get("runtime_added_tool_names", [])) - set(bot._toolset.tools)
        if missing:
            warnings.warn(
                f"load_history() bundle references runtime-added tool(s) not "
                f"present after reload: {sorted(missing)}. Pass them again via "
                f"tools=[...] or call add_tool() to restore them.",
                stacklevel=2,
            )
        return bot

    @property
    def store(self) -> ResultStore:
        return self._store

chat async 

chat(message: str, *, extra_tools: list[Any] | None = None) -> BotResponse

Send a message and accumulate history (multi-turn entry point).

Source code in aitaem/agent/base.py 
107
108
109
110
111
112
113
114
115
116
async def chat(
    self,
    message: str,
    *,
    extra_tools: list[Any] | None = None,
) -> BotResponse:
    """Send a message and accumulate history (multi-turn entry point)."""
    raise NotImplementedError(
        "chat() must be implemented by the convenience bot subclass."
    )

ask async 

ask(message: str, *, extra_tools: list[Any] | None = None) -> BotResponse

Send a single-turn message without accumulating history.

Source code in aitaem/agent/base.py 
118
119
120
121
122
123
124
125
126
127
async def ask(
    self,
    message: str,
    *,
    extra_tools: list[Any] | None = None,
) -> BotResponse:
    """Send a single-turn message without accumulating history."""
    raise NotImplementedError(
        "ask() must be implemented by the convenience bot subclass."
    )

add_tool 

add_tool(tool: Any) -> None

Add a tool to this bot's persistent tool set at runtime.

Takes effect on the next chat()/ask() call. Mutations during an in-progress run() are undefined.

Source code in aitaem/agent/base.py 
129
130
131
132
133
134
135
136
137
def add_tool(self, tool: Any) -> None:
    """Add a tool to this bot's persistent tool set at runtime.

    Takes effect on the next chat()/ask() call. Mutations during an
    in-progress run() are undefined.
    """
    before = set(self._toolset.tools)
    _register_tool(self._toolset, tool)
    self._runtime_added_tool_names.extend(sorted(set(self._toolset.tools) - before))

add_bot 

add_bot(bot: Bot) -> None

Register another bot as a tool (sugar for add_tool(other.as_tool())).

Source code in aitaem/agent/base.py 
139
140
141
def add_bot(self, bot: Bot) -> None:
    """Register another bot as a tool (sugar for add_tool(other.as_tool()))."""
    self.add_tool(bot.as_tool())

as_tool 

as_tool() -> Any

Return a pydantic-ai Tool that wraps this bot's ask() method.

Source code in aitaem/agent/base.py 
143
144
145
def as_tool(self) -> Any:
    """Return a pydantic-ai Tool that wraps this bot's ask() method."""
    raise NotImplementedError("as_tool() implemented in Phase 5.")

get_result 

get_result(result_id: str) -> ResultEntry

Retrieve a stored computation result by ID.

Source code in aitaem/agent/base.py 
147
148
149
def get_result(self, result_id: str) -> ResultEntry:
    """Retrieve a stored computation result by ID."""
    return self._store.get(result_id)

dump_history 

dump_history() -> dict[str, Any]

Serialize conversation history and result artifacts to a JSON-safe dict.

The returned dict is JSON-serializable (suitable for json.dumps). Ibis refs are not serialized; only Arrow artifacts are preserved. Reload with load_history() to restore history with Arrow artifacts available via get_result(), but get_ibis() will return None on restored entries.

Tools added at runtime via add_tool() are NOT restored by load_history() — their names are recorded in the bundle so load_history() can warn if they're missing after reload, but the callables themselves aren't portably serializable. Pass them again via tools=[...] or call add_tool() on the reloaded bot to restore them.

Source code in aitaem/agent/base.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def dump_history(self) -> dict[str, Any]:
    """Serialize conversation history and result artifacts to a JSON-safe dict.

    The returned dict is JSON-serializable (suitable for json.dumps). Ibis refs
    are not serialized; only Arrow artifacts are preserved. Reload with
    load_history() to restore history with Arrow artifacts available via
    get_result(), but get_ibis() will return None on restored entries.

    Tools added at runtime via add_tool() are NOT restored by load_history()
    — their names are recorded in the bundle so load_history() can warn if
    they're missing after reload, but the callables themselves aren't
    portably serializable. Pass them again via tools=[...] or call
    add_tool() on the reloaded bot to restore them.
    """
    from aitaem.agent.history import make_bundle

    return make_bundle(
        self._message_history, self._store, self._runtime_added_tool_names
    )

load_history classmethod 

load_history(data: dict[str, Any], **kwargs: Any) -> Bot

Construct a new bot pre-loaded with a serialized history bundle.

Parameters:

Name Type Description Default
data dict[str, Any]

Bundle returned by dump_history() on a prior bot instance.

 required
**kwargs Any

Constructor arguments for the concrete Bot subclass.

 {}

Returns:

Type Description
Bot

A new bot instance with _message_history and store populated from
Bot

data. The bot's _agent is rebuilt fresh from **kwargs.

Warns (UserWarning) if the bundle references tools added via add_tool() on the original bot that are not present on the reloaded bot — pass them again via tools=[...] in **kwargs, or call add_tool() after reload, to silence the warning.

Source code in aitaem/agent/base.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
@classmethod
def load_history(cls, data: dict[str, Any], **kwargs: Any) -> Bot:
    """Construct a new bot pre-loaded with a serialized history bundle.

    Args:
        data: Bundle returned by dump_history() on a prior bot instance.
        **kwargs: Constructor arguments for the concrete Bot subclass.

    Returns:
        A new bot instance with _message_history and store populated from
        data. The bot's _agent is rebuilt fresh from **kwargs.

    Warns (UserWarning) if the bundle references tools added via
    add_tool() on the original bot that are not present on the reloaded
    bot — pass them again via tools=[...] in **kwargs, or call add_tool()
    after reload, to silence the warning.
    """
    from aitaem.agent.history import load_bundle

    bot = cls(**kwargs)
    bot._message_history = load_bundle(data, bot._store)
    missing = set(data.get("runtime_added_tool_names", [])) - set(bot._toolset.tools)
    if missing:
        warnings.warn(
            f"load_history() bundle references runtime-added tool(s) not "
            f"present after reload: {sorted(missing)}. Pass them again via "
            f"tools=[...] or call add_tool() to restore them.",
            stacklevel=2,
        )
    return bot

BotResponse

aitaem.agent.response.BotResponse

Bases: BaseModel, Generic[PayloadT]

The response returned by every Bot method — status, narrative, trace, and an optional typed payload.

Source code in aitaem/agent/response.py 
12
13
14
15
16
17
18
19
20
21
22
class BotResponse(BaseModel, Generic[PayloadT]):
    """The response returned by every Bot method — status, narrative, trace, and an
    optional typed payload."""

    model_config = ConfigDict(frozen=True)

    status: Status
    narrative: str
    trace: RunTrace
    reason: str | None = None
    payload: PayloadT | None = None

Status

aitaem.agent.response.Status

Bases: str, Enum

Outcome of a bot response: ok, empty, refused, or error.

refused is the general "no spec precisely answers — don't substitute an approximation" outcome (the Metric Precision Rule), distinct from error (a tool or the run itself failed) and empty (ran fine, no results).

Source code in aitaem/agent/trace.py 
11
12
13
14
15
16
17
18
19
20
21
22
class Status(str, Enum):
    """Outcome of a bot response: ok, empty, refused, or error.

    `refused` is the general "no spec precisely answers — don't substitute an
    approximation" outcome (the Metric Precision Rule), distinct from `error`
    (a tool or the run itself failed) and `empty` (ran fine, no results).
    """

    ok = "ok"
    empty = "empty"
    refused = "refused"
    error = "error"

RunTrace

aitaem.agent.trace.RunTrace

Bases: BaseModel

Aggregated, eval-friendly trace of one bot run — tool calls, usage, and timing.

Source code in aitaem/agent/trace.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
class RunTrace(BaseModel):
    """Aggregated, eval-friendly trace of one bot run — tool calls, usage, and timing."""

    model_config = ConfigDict(frozen=True)

    run_id: str
    conversation_id: str
    timestamp: datetime
    tool_calls: list[ToolCall]
    usage: Usage
    traceparent: str | None = None
    duration_ms: float = 0.0
    error: str | None = None

ToolCall

aitaem.agent.trace.ToolCall

Bases: BaseModel

One tool invocation within a run, with its arguments, outcome, and an LLM-facing summary (never the raw result — see result_id for that).

Source code in aitaem/agent/trace.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
class ToolCall(BaseModel):
    """One tool invocation within a run, with its arguments, outcome, and an
    LLM-facing summary (never the raw result — see `result_id` for that)."""

    model_config = ConfigDict(frozen=True)

    tool_call_id: str
    name: str
    args: dict[str, Any]
    result_id: str | None = None
    llm_summary: str | None = None  # compact snippet only — never raw result data
    success: bool = True
    duration_ms: float | None = None

Usage

aitaem.agent.trace.Usage

Bases: BaseModel

Token and tool-call usage counters for one bot run.

Source code in aitaem/agent/trace.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
class Usage(BaseModel):
    """Token and tool-call usage counters for one bot run."""

    model_config = ConfigDict(frozen=True)

    requests: int = 0
    tool_calls: int = 0
    input_tokens: int = 0
    output_tokens: int = 0
    cache_read_tokens: int = 0
    cache_write_tokens: int = 0

    @computed_field  # type: ignore[prop-decorator]
    @property
    def total_tokens(self) -> int:
        return self.input_tokens + self.output_tokens

    @classmethod
    def from_run_usage(cls, ru: Any) -> Usage:
        return cls(
            requests=ru.requests,
            tool_calls=ru.tool_calls,
            input_tokens=ru.input_tokens,
            output_tokens=ru.output_tokens,
            cache_read_tokens=ru.cache_read_tokens,
            cache_write_tokens=ru.cache_write_tokens,
        )

ResultStore

aitaem.agent.store.ResultStore

Session-scoped store for computation results.

Supports two entry kinds: tabular (Arrow + Ibis ref) and text (string + content_type). Use store_tabular() / get_tabular() for computation results; store_text() / get_text() for serialized artifacts such as validated YAML specs. The generic get() returns the union and is useful when the caller handles either kind.

Source code in aitaem/agent/store.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
class ResultStore:
    """Session-scoped store for computation results.

    Supports two entry kinds: tabular (Arrow + Ibis ref) and text (string + content_type).
    Use store_tabular() / get_tabular() for computation results; store_text() / get_text()
    for serialized artifacts such as validated YAML specs. The generic get() returns the
    union and is useful when the caller handles either kind.
    """

    def __init__(self) -> None:
        self._entries: dict[str, TabularEntry | TextEntry] = {}

    def store_tabular(
        self,
        arrow: pa.Table | None,
        ibis_ref: Any | None,
        metadata: dict[str, Any] | None = None,
    ) -> str:
        result_id = str(uuid.uuid4())
        self._entries[result_id] = TabularEntry(
            result_id=result_id,
            arrow=arrow,
            ibis_ref=ibis_ref,
            metadata=metadata or {},
        )
        return result_id

    def store_text(
        self,
        text: str,
        content_type: str,
        metadata: dict[str, Any] | None = None,
    ) -> str:
        result_id = str(uuid.uuid4())
        self._entries[result_id] = TextEntry(
            result_id=result_id,
            text=text,
            content_type=content_type,
            metadata=metadata or {},
        )
        return result_id

    def get(self, result_id: str) -> TabularEntry | TextEntry:
        try:
            return self._entries[result_id]
        except KeyError:
            raise KeyError(f"No result with id={result_id!r}")

    def get_tabular(self, result_id: str) -> TabularEntry:
        entry = self.get(result_id)
        if not isinstance(entry, TabularEntry):
            raise WrongEntryKindError(
                f"Entry {result_id!r} has kind={entry.kind!r}, expected 'tabular'."
            )
        return entry

    def get_text(self, result_id: str) -> TextEntry:
        entry = self.get(result_id)
        if not isinstance(entry, TextEntry):
            raise WrongEntryKindError(
                f"Entry {result_id!r} has kind={entry.kind!r}, expected 'text'."
            )
        return entry

    def get_ibis(self, result_id: str) -> Any | None:
        return self.get_tabular(result_id).ibis_ref

    def get_arrow(self, result_id: str) -> pa.Table | None:
        return self.get_tabular(result_id).arrow

    def invalidate_all_ibis_refs(self) -> None:
        for entry in self._entries.values():
            if isinstance(entry, TabularEntry):
                entry.invalidate_ibis_ref()

    def __len__(self) -> int:
        return len(self._entries)

    def ids(self) -> list[str]:
        return list(self._entries.keys())

ResultEntry

aitaem.agent.store.ResultEntry module-attribute 

ResultEntry = TabularEntry | TextEntry

TabularEntry

aitaem.agent.store.TabularEntry

Bases: _EntryBase

Tabular result entry holding an Arrow artifact and optional live ibis ref.

Source code in aitaem/agent/store.py 
27
28
29
30
31
32
33
34
35
class TabularEntry(_EntryBase):
    """Tabular result entry holding an Arrow artifact and optional live ibis ref."""

    kind: Literal["tabular"] = "tabular"
    arrow: pa.Table | None = None
    ibis_ref: Any | None = None

    def invalidate_ibis_ref(self) -> None:
        self.ibis_ref = None

TextEntry

aitaem.agent.store.TextEntry

Bases: _EntryBase

Text artifact entry (e.g. validated YAML spec, JSON config).

Source code in aitaem/agent/store.py 
38
39
40
41
42
43
class TextEntry(_EntryBase):
    """Text artifact entry (e.g. validated YAML spec, JSON config)."""

    kind: Literal["text"] = "text"
    text: str
    content_type: str

WrongEntryKindError

aitaem.agent.store.WrongEntryKindError

Bases: AitaemError

Raised when a ResultStore entry is accessed with the wrong kind getter.

Source code in aitaem/agent/store.py 
13
14
class WrongEntryKindError(AitaemError):
    """Raised when a ResultStore entry is accessed with the wrong kind getter."""

QueryBot

QueryBot

aitaem.agent.query_bot.QueryBot

Bases: Bot

Convenience bot for answering natural-language questions against a metric catalog.

Tools create a MetricCompute instance per call from the held spec_cache and connection_manager. Artifacts are written to the bot's ResultStore; callers dereference via get_result(result_id).

Construction

bot = QueryBot( model="anthropic:claude-sonnet-4-6", spec_cache=my_spec_cache, connection_manager=my_connection_manager, ) response = await bot.chat("What was Q4 revenue by region?")

Multi-provider

Use model strings supported by pydantic-ai, e.g. "openai:gpt-4o". For testing, pass a FunctionModel or TestModel instance directly.

tenant_id

Optional per-tenant identifier for OpenAI prompt-cache routing. When omitted, a fingerprint of the spec_cache's visible catalog is used, which naturally separates RBAC-differentiated permission sets.

Source code in aitaem/agent/query_bot.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
class QueryBot(Bot):
    """Convenience bot for answering natural-language questions against a metric catalog.

    Tools create a MetricCompute instance per call from the held spec_cache and
    connection_manager. Artifacts are written to the bot's ResultStore; callers
    dereference via get_result(result_id).

    Construction:
        bot = QueryBot(
            model="anthropic:claude-sonnet-4-6",
            spec_cache=my_spec_cache,
            connection_manager=my_connection_manager,
        )
        response = await bot.chat("What was Q4 revenue by region?")

    Multi-provider:
        Use model strings supported by pydantic-ai, e.g. "openai:gpt-4o".
        For testing, pass a FunctionModel or TestModel instance directly.

    tenant_id:
        Optional per-tenant identifier for OpenAI prompt-cache routing.
        When omitted, a fingerprint of the spec_cache's visible catalog is used,
        which naturally separates RBAC-differentiated permission sets.
    """

    def __init__(
        self,
        *,
        model: Any,
        spec_cache: Any,
        connection_manager: Any,
        tenant_id: str | None = None,
        tools: list[Any] | None = None,
    ) -> None:
        # Set bot-specific resources BEFORE super().__init__() — _build_agent()
        # is called inside super().__init__() and needs these attributes.
        self._spec_cache = spec_cache
        self._connection_manager = connection_manager
        self._tenant_id = tenant_id
        super().__init__(model=model, tools=tools)
        self._conversation_id: str | None = None

    def _build_agent(self) -> Any:
        from pydantic_ai import Agent
        from pydantic_ai.toolsets import FunctionToolset
        from pydantic_ai.capabilities import ReinjectSystemPrompt

        toolset = FunctionToolset()
        toolset.add_function(record_intent)        # Step 1
        toolset.add_function(resolve_intent)       # Step 2
        toolset.add_function(compute_metrics)      # Step 3
        toolset.add_function(rank_by_value)
        toolset.add_function(filter_by_threshold)
        toolset.add_function(distribution_summary)
        toolset.add_function(period_over_period)
        toolset.add_function(contribution_share)

        for tool in self._tools:
            _register_tool(toolset, tool)
        self._toolset = toolset

        # Static instructions: Layers A + B combined.
        # These become InstructionPart(dynamic=False) and are cached at the
        # provider-appropriate breakpoint (see _provider_cache_config above).
        static_instructions = _build_layer_a() + "\n\n" + _build_layer_b(self._spec_cache)

        # Derive a stable routing key for OpenAI. Explicit tenant_id wins; fall back
        # to _permission_fingerprint so single-tenant installs require zero config and
        # RBAC-differentiated users naturally land in separate routing lanes.
        tenant_id = self._tenant_id or _permission_fingerprint(self._spec_cache)

        agent = Agent(  # type: ignore[call-overload]
            model=self._model,
            deps_type=QueryDeps,
            output_type=QueryOutput,
            toolsets=[toolset],
            instructions=static_instructions,
            # anthropic_cache_instructions: verified against pydantic-ai 2.2.0's
            # anthropic adapter. InstructionPart.sorted() sorts static (dynamic=False)
            # before dynamic parts; the adapter then sets
            # cache_block_idx = num_prefix_blocks + num_static - 1, placing the
            # cache_control breakpoint after the last static block (Layer B). Layer C
            # follows as a dynamic block and is NOT cached. Other providers ignore this.
            model_settings=_provider_cache_config(self._model, tenant_id),
            capabilities=[ReinjectSystemPrompt(replace_existing=True)],
        )

        # Layer C: per-turn date context (dynamic=True → NOT cached).
        # Registered here, after the agent is built, so it captures today's date on each run.
        @agent.instructions
        def _layer_c() -> str:
            from datetime import date
            return (
                f"# ─── Layer C: per-turn context ─────────────────────────────────────────────\n\n"
                f"Today is {date.today().isoformat()}. Use it to resolve relative time references "
                f'("last month", "recently", "May") into concrete time_window values before '
                f"calling record_intent."
            )

        return agent

    async def chat(
        self,
        message: str,
        *,
        extra_tools: list[Any] | None = None,
    ) -> QueryResponse:
        """Send a message in multi-turn mode. Accumulates history on the bot.

        Always returns a QueryResponse — exceptions from the agent run are caught
        and surfaced as status=error rather than propagated raw.
        """
        from datetime import datetime, timezone
        from aitaem.agent.trace import assemble_trace

        run_start = datetime.now(timezone.utc)
        deps = QueryDeps(
            spec_cache=self._spec_cache,
            connection_manager=self._connection_manager,
            store=self._store,
        )
        try:
            run_kwargs: dict[str, Any] = {
                "message_history": self._message_history,
                "deps": deps,
            }
            extra_toolset = _build_extra_toolset(extra_tools)
            if extra_toolset is not None:
                run_kwargs["toolsets"] = [extra_toolset]
            result = await self._agent.run(message, **run_kwargs)
            self._message_history = result.all_messages()
            output = cast(QueryOutput, result.output)
            trace = assemble_trace(result, run_start)
            self._conversation_id = trace.conversation_id
            payload = QueryBot._assemble_payload(output, trace)
            return QueryResponse(
                status=output.status,
                narrative=output.narrative,
                trace=trace,
                reason=output.reason,
                payload=payload,
            )
        except Exception as exc:
            return QueryBot._error_response(exc, run_start, self._conversation_id)

    async def ask(
        self,
        message: str,
        *,
        extra_tools: list[Any] | None = None,
    ) -> QueryResponse:
        """Send a single-turn message. Does NOT accumulate history.

        Always returns a QueryResponse — exceptions from the agent run are caught
        and surfaced as status=error rather than propagated raw.
        """
        from datetime import datetime, timezone
        from aitaem.agent.trace import assemble_trace

        run_start = datetime.now(timezone.utc)
        deps = QueryDeps(
            spec_cache=self._spec_cache,
            connection_manager=self._connection_manager,
            store=self._store,
        )
        try:
            run_kwargs: dict[str, Any] = {"deps": deps}
            extra_toolset = _build_extra_toolset(extra_tools)
            if extra_toolset is not None:
                run_kwargs["toolsets"] = [extra_toolset]
            result = await self._agent.run(message, **run_kwargs)
            output = cast(QueryOutput, result.output)
            trace = assemble_trace(result, run_start)
            self._conversation_id = trace.conversation_id
            payload = QueryBot._assemble_payload(output, trace)
            return QueryResponse(
                status=output.status,
                narrative=output.narrative,
                trace=trace,
                reason=output.reason,
                payload=payload,
            )
        except Exception as exc:
            return QueryBot._error_response(exc, run_start, self._conversation_id)

    @staticmethod
    def _error_response(
        exc: Exception, run_start: Any, conversation_id: str | None
    ) -> QueryResponse:
        """Build a status=error QueryResponse when _agent.run() raises."""
        import uuid
        from aitaem.agent.trace import RunTrace, Usage

        trace = RunTrace(
            run_id=str(uuid.uuid4()),
            conversation_id=conversation_id or str(uuid.uuid4()),
            timestamp=run_start,
            tool_calls=[],
            usage=Usage(),
            error=f"{type(exc).__name__}: {exc}",
        )
        return QueryResponse(
            status=Status.error,
            narrative="The request could not be completed due to an unexpected error.",
            trace=trace,
            reason=str(exc),
            payload=QueryPayload(
                result_ids=[], primary_result_id=None,
                metrics_used=[], slices_used=[], segment_used=None,
                time_window=None, period_type="all_time", by_entity=None,
            ),
        )

    @staticmethod
    def _assemble_payload(output: QueryOutput, trace: Any) -> QueryPayload:
        """Assemble QueryPayload from the LLM's QueryOutput and the turn trace.

        Reads payload_summary from each tool's llm_summary (JSON-serialized
        ToolResult). Tool-agnostic: no per-tool field access needed.

        Aggregation rules across multiple tool calls:
          list fields  — union with deduplication, order of first appearance
          scalar fields — first-write wins (first call that sets a field governs)
        """
        primary_result_id = output.result_ids[0] if output.result_ids else None
        metrics_used: list[str] = []
        slices_used: list[str] = []
        seen_metrics: set[str] = set()
        seen_slices: set[str] = set()
        segment_used: str | None = None
        time_window: tuple[str, str] | None = None
        period_type: str | None = None
        by_entity: str | None = None
        format_hints: dict[str, str] = {}
        sample: list[dict[str, Any]] | None = None

        for tc in trace.tool_calls:
            if not tc.llm_summary:
                continue
            try:
                summary = json.loads(tc.llm_summary)
            except (ValueError, TypeError):
                continue
            ps = summary.get("payload_summary")
            if not ps:
                continue
            for m in ps.get("metrics_used") or []:
                if m not in seen_metrics:
                    seen_metrics.add(m)
                    metrics_used.append(m)
            for s in ps.get("slices_used") or []:
                if s not in seen_slices:
                    seen_slices.add(s)
                    slices_used.append(s)
            if segment_used is None and ps.get("segment_used"):
                segment_used = ps["segment_used"]
            if time_window is None and ps.get("time_window"):
                tw = ps["time_window"]
                time_window = (tw[0], tw[1]) if isinstance(tw, (list, tuple)) and len(tw) == 2 else None
            if period_type is None and ps.get("period_type"):
                period_type = ps["period_type"]
            if by_entity is None and ps.get("by_entity"):
                by_entity = ps["by_entity"]
            for metric, fmt in (ps.get("format_hints") or {}).items():
                if metric not in format_hints:
                    format_hints[metric] = fmt
            if sample is None and primary_result_id and ps.get("result_id") == primary_result_id:
                raw = ps.get("sample")
                sample = raw if isinstance(raw, list) else None

        return QueryPayload(
            result_ids=output.result_ids,
            primary_result_id=primary_result_id,
            metrics_used=metrics_used,
            slices_used=slices_used,
            segment_used=segment_used,
            time_window=time_window,
            period_type=period_type or "all_time",
            by_entity=by_entity,
            format_hints=format_hints,
            sample=sample,
        )

chat async 

chat(message: str, *, extra_tools: list[Any] | None = None) -> QueryResponse

Send a message in multi-turn mode. Accumulates history on the bot.

Always returns a QueryResponse — exceptions from the agent run are caught and surfaced as status=error rather than propagated raw.

Source code in aitaem/agent/query_bot.py 
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
async def chat(
    self,
    message: str,
    *,
    extra_tools: list[Any] | None = None,
) -> QueryResponse:
    """Send a message in multi-turn mode. Accumulates history on the bot.

    Always returns a QueryResponse — exceptions from the agent run are caught
    and surfaced as status=error rather than propagated raw.
    """
    from datetime import datetime, timezone
    from aitaem.agent.trace import assemble_trace

    run_start = datetime.now(timezone.utc)
    deps = QueryDeps(
        spec_cache=self._spec_cache,
        connection_manager=self._connection_manager,
        store=self._store,
    )
    try:
        run_kwargs: dict[str, Any] = {
            "message_history": self._message_history,
            "deps": deps,
        }
        extra_toolset = _build_extra_toolset(extra_tools)
        if extra_toolset is not None:
            run_kwargs["toolsets"] = [extra_toolset]
        result = await self._agent.run(message, **run_kwargs)
        self._message_history = result.all_messages()
        output = cast(QueryOutput, result.output)
        trace = assemble_trace(result, run_start)
        self._conversation_id = trace.conversation_id
        payload = QueryBot._assemble_payload(output, trace)
        return QueryResponse(
            status=output.status,
            narrative=output.narrative,
            trace=trace,
            reason=output.reason,
            payload=payload,
        )
    except Exception as exc:
        return QueryBot._error_response(exc, run_start, self._conversation_id)

ask async 

ask(message: str, *, extra_tools: list[Any] | None = None) -> QueryResponse

Send a single-turn message. Does NOT accumulate history.

Always returns a QueryResponse — exceptions from the agent run are caught and surfaced as status=error rather than propagated raw.

Source code in aitaem/agent/query_bot.py 
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
async def ask(
    self,
    message: str,
    *,
    extra_tools: list[Any] | None = None,
) -> QueryResponse:
    """Send a single-turn message. Does NOT accumulate history.

    Always returns a QueryResponse — exceptions from the agent run are caught
    and surfaced as status=error rather than propagated raw.
    """
    from datetime import datetime, timezone
    from aitaem.agent.trace import assemble_trace

    run_start = datetime.now(timezone.utc)
    deps = QueryDeps(
        spec_cache=self._spec_cache,
        connection_manager=self._connection_manager,
        store=self._store,
    )
    try:
        run_kwargs: dict[str, Any] = {"deps": deps}
        extra_toolset = _build_extra_toolset(extra_tools)
        if extra_toolset is not None:
            run_kwargs["toolsets"] = [extra_toolset]
        result = await self._agent.run(message, **run_kwargs)
        output = cast(QueryOutput, result.output)
        trace = assemble_trace(result, run_start)
        self._conversation_id = trace.conversation_id
        payload = QueryBot._assemble_payload(output, trace)
        return QueryResponse(
            status=output.status,
            narrative=output.narrative,
            trace=trace,
            reason=output.reason,
            payload=payload,
        )
    except Exception as exc:
        return QueryBot._error_response(exc, run_start, self._conversation_id)

QueryResponse

aitaem.agent.query_bot.QueryResponse

Bases: BotResponse[QueryPayload]

Concrete response type for QueryBot — narrows BotResponse's generic payload.

Source code in aitaem/agent/query_bot.py 
39
40
class QueryResponse(BotResponse[QueryPayload]):
    """Concrete response type for QueryBot — narrows BotResponse's generic payload."""

QueryPayload

aitaem.agent.query_types.QueryPayload

Bases: BaseModel

Metadata assembled by QueryBot from QueryOutput and the turn trace.

Source code in aitaem/agent/query_types.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
class QueryPayload(BaseModel):
    """Metadata assembled by QueryBot from QueryOutput and the turn trace."""
    model_config = ConfigDict(frozen=True)

    result_ids: list[str]
    primary_result_id: str | None   # first entry of result_ids, or None
    metrics_used: list[str]
    slices_used: list[str]
    segment_used: str | None
    time_window: tuple[str, str] | None
    period_type: str
    by_entity: str | None
    format_hints: dict[str, str] = Field(
        default_factory=dict,
        description=(
            "metric_name → format string (e.g. 'percentage', 'currency:USD'). "
            "Callers use this to render metric values correctly."
        ),
    )
    sample: list[dict[str, Any]] | None = Field(
        default=None,
        description=(
            "Up to 5 rows from the primary result, with Python-native values. "
            "None when there is no primary result."
        ),
    )

MetricIntent

aitaem.agent.query_types.MetricIntent dataclass

Structured interpretation of one metric the user is asking about.

Produced by record_intent and stored in QueryDeps.intents. One intent per metric; multi-metric questions produce multiple intents.

Source code in aitaem/agent/query_types.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
@dataclass
class MetricIntent:
    """Structured interpretation of one metric the user is asking about.

    Produced by record_intent and stored in QueryDeps.intents.
    One intent per metric; multi-metric questions produce multiple intents.
    """
    metric_concept: str                          # free-text LLM interpretation
    scope: Literal["overall", "subset"]
    subset_description: str | None = None        # prose description of the subset
    slice_type: str | None = None                # proposed slice spec name (breakdown)
    slice_value: str | None = None               # specific filter value, e.g. "US"
    segment_name: str | None = None              # proposed segment spec name
    segment_value: str | None = None             # specific segment filter value
    period_type: str = "all_time"
    time_window: tuple[str, str] | None = None   # (start_iso, end_iso)
    by_entity: str | None = None

ResolvedSpec

aitaem.agent.query_types.ResolvedSpec dataclass

Validated compute parameters keyed by spec_token in QueryDeps.spec_registry.

Constructed by resolve_intent when SpecResolver confirms an exact match. Consumed by compute_metrics(spec_token) to reconstruct MetricCompute arguments.

Source code in aitaem/agent/query_types.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
@dataclass
class ResolvedSpec:
    """Validated compute parameters keyed by spec_token in QueryDeps.spec_registry.

    Constructed by resolve_intent when SpecResolver confirms an exact match.
    Consumed by compute_metrics(spec_token) to reconstruct MetricCompute arguments.
    """
    metric_name: str
    slice_specs: list[str]           # validated slice spec names
    segment_spec: str | None         # validated segment spec name
    period_type: str
    time_window: tuple[str, str] | None
    by_entity: str | None
    intent_slice_value: str | None   # from MetricIntent; for trace only
    intent_segment_value: str | None

ExactMatch

aitaem.agent.query_types.ExactMatch

Bases: BaseModel

Minted only when SpecResolver confirms a valid proposal.

Source code in aitaem/agent/query_types.py 
65
66
67
68
69
70
class ExactMatch(BaseModel):
    """Minted only when SpecResolver confirms a valid proposal."""
    spec_token: str
    metric_name: str
    slices: list[str]
    segment: str | None

NearMiss

aitaem.agent.query_types.NearMiss

Bases: BaseModel

A catalog entry that resolve_intent considered but rejected, with the reason why.

Source code in aitaem/agent/query_types.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
class NearMiss(BaseModel):
    """A catalog entry that resolve_intent considered but rejected, with the reason why."""

    name: str
    why_not: Literal[
        "unknown_metric",
        "scope_mismatch", "wrong_dimension_kind",
        "unknown_slice", "unknown_segment",
        "unsupported_by_entity", "unsupported_period_type",
    ]
    suggestions: list[str] = []
    """Catalog names close to `name`. Non-empty only when why_not='unknown_metric'.
    Populated via difflib.get_close_matches (cutoff=0.75) for typo correction.
    Empty for all other why_not reasons."""

suggestions class-attribute instance-attribute 

suggestions: list[str] = []

Catalog names close to name. Non-empty only when why_not='unknown_metric'. Populated via difflib.get_close_matches (cutoff=0.75) for typo correction. Empty for all other why_not reasons.

SpecMatchResult

aitaem.agent.query_types.SpecMatchResult

Bases: BaseModel

Returned to the LLM by resolve_intent.

If exact_match is not None: the LLM proceeds to compute_metrics(spec_token). If exact_match is None: the LLM must produce status=refused and cite near_misses.

Source code in aitaem/agent/query_types.py 
89
90
91
92
93
94
95
96
class SpecMatchResult(BaseModel):
    """Returned to the LLM by resolve_intent.

    If exact_match is not None: the LLM proceeds to compute_metrics(spec_token).
    If exact_match is None: the LLM must produce status=refused and cite near_misses.
    """
    exact_match: ExactMatch | None
    near_misses: list[NearMiss]

RecordIntentResult

aitaem.agent.query_types.RecordIntentResult

Bases: BaseModel

Returned by record_intent. The intent_id is used in the resolve_intent call.

Source code in aitaem/agent/query_types.py 
 99
100
101
class RecordIntentResult(BaseModel):
    """Returned by record_intent. The intent_id is used in the resolve_intent call."""
    intent_id: int

ResolveIntentResult

aitaem.agent.query_types.ResolveIntentResult

Bases: BaseModel

Returned by resolve_intent. Wraps SpecMatchResult for the LLM.

Source code in aitaem/agent/query_types.py 
104
105
106
107
class ResolveIntentResult(BaseModel):
    """Returned by resolve_intent. Wraps SpecMatchResult for the LLM."""
    exact_match: ExactMatch | None
    near_misses: list[NearMiss]

SpecResolver

aitaem.agent.resolver.SpecResolver

Deterministic v0 catalog validator.

v0 → v1 swap point: the interface (resolve method signature and return type) is stable. Only the body changes in v1 (dict lookup → RAG retrieval + deterministic filter).

Source code in aitaem/agent/resolver.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class SpecResolver:
    """Deterministic v0 catalog validator.

    v0 → v1 swap point: the interface (resolve method signature and return type) is
    stable. Only the body changes in v1 (dict lookup → RAG retrieval + deterministic filter).
    """

    def resolve(
        self,
        intent: MetricIntent,
        proposed_metric_name: str,
        proposed_slices: list[str],
        proposed_segment: str | None,
        spec_cache: Any,  # aitaem.SpecCache
    ) -> SpecMatchResult:
        """Validate the proposed names against the catalog.

        Returns SpecMatchResult with exact_match set if all validations pass.
        The spec_token inside exact_match is left empty (""); the caller (resolve_intent
        tool) mints and fills the token after this method returns.
        """
        near_misses: list[NearMiss] = []

        # scope_mismatch is deliberately NOT checked in v0. MetricSpec has no
        # "scope" flag, so the resolver cannot distinguish an inherently-scoped
        # metric (e.g. `ctr_conversion_ads`) from an overall metric proposed
        # for a subset intent. The LLM's metric selection is trusted. Revisit
        # if a future MetricSpec field marks scope explicitly.

        # ── 1. Validate metric name ──────────────────────────────────────────
        metric_spec = spec_cache.metrics.get(proposed_metric_name)
        if metric_spec is None:
            # Unknown metric — can't validate slices/segment without the spec, so return early.
            # Populate suggestions via fuzzy match to help the LLM surface typo corrections.
            suggestions = difflib.get_close_matches(
                proposed_metric_name, spec_cache.metrics.keys(), n=3, cutoff=0.75
            )
            return SpecMatchResult(
                exact_match=None,
                near_misses=near_misses + [
                    NearMiss(name=proposed_metric_name, why_not="unknown_metric", suggestions=suggestions)
                ],
            )

        # ── 2. Validate slices ───────────────────────────────────────────────
        for slice_name in proposed_slices:
            if slice_name in spec_cache.slices:
                pass  # valid
            elif slice_name in spec_cache.segments:
                near_misses.append(NearMiss(name=slice_name, why_not="wrong_dimension_kind"))
            else:
                near_misses.append(NearMiss(name=slice_name, why_not="unknown_slice"))

        # ── 3. Validate segment ──────────────────────────────────────────────
        if proposed_segment is not None:
            if proposed_segment in spec_cache.segments:
                pass  # valid
            elif proposed_segment in spec_cache.slices:
                near_misses.append(NearMiss(name=proposed_segment, why_not="wrong_dimension_kind"))
            else:
                near_misses.append(NearMiss(name=proposed_segment, why_not="unknown_segment"))

        # ── 4. Validate by_entity ────────────────────────────────────────────
        if intent.by_entity is not None:
            entities = metric_spec.entities or []
            if intent.by_entity not in entities:
                near_misses.append(NearMiss(
                    name=proposed_metric_name, why_not="unsupported_by_entity"
                ))

        # ── 5. Validate period_type ──────────────────────────────────────────
        if intent.period_type != "all_time" and not metric_spec.timestamp_col:
            near_misses.append(NearMiss(
                name=proposed_metric_name, why_not="unsupported_period_type"
            ))

        # ── Result ───────────────────────────────────────────────────────────
        if near_misses:
            return SpecMatchResult(exact_match=None, near_misses=near_misses)

        return SpecMatchResult(
            exact_match=ExactMatch(
                spec_token="",  # caller (resolve_intent tool) mints and fills this
                metric_name=proposed_metric_name,
                slices=proposed_slices,
                segment=proposed_segment,
            ),
            near_misses=[],
        )

resolve 

resolve(intent: MetricIntent, proposed_metric_name: str, proposed_slices: list[str], proposed_segment: str | None, spec_cache: Any) -> SpecMatchResult

Validate the proposed names against the catalog.

Returns SpecMatchResult with exact_match set if all validations pass. The spec_token inside exact_match is left empty (""); the caller (resolve_intent tool) mints and fills the token after this method returns.

Source code in aitaem/agent/resolver.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def resolve(
    self,
    intent: MetricIntent,
    proposed_metric_name: str,
    proposed_slices: list[str],
    proposed_segment: str | None,
    spec_cache: Any,  # aitaem.SpecCache
) -> SpecMatchResult:
    """Validate the proposed names against the catalog.

    Returns SpecMatchResult with exact_match set if all validations pass.
    The spec_token inside exact_match is left empty (""); the caller (resolve_intent
    tool) mints and fills the token after this method returns.
    """
    near_misses: list[NearMiss] = []

    # scope_mismatch is deliberately NOT checked in v0. MetricSpec has no
    # "scope" flag, so the resolver cannot distinguish an inherently-scoped
    # metric (e.g. `ctr_conversion_ads`) from an overall metric proposed
    # for a subset intent. The LLM's metric selection is trusted. Revisit
    # if a future MetricSpec field marks scope explicitly.

    # ── 1. Validate metric name ──────────────────────────────────────────
    metric_spec = spec_cache.metrics.get(proposed_metric_name)
    if metric_spec is None:
        # Unknown metric — can't validate slices/segment without the spec, so return early.
        # Populate suggestions via fuzzy match to help the LLM surface typo corrections.
        suggestions = difflib.get_close_matches(
            proposed_metric_name, spec_cache.metrics.keys(), n=3, cutoff=0.75
        )
        return SpecMatchResult(
            exact_match=None,
            near_misses=near_misses + [
                NearMiss(name=proposed_metric_name, why_not="unknown_metric", suggestions=suggestions)
            ],
        )

    # ── 2. Validate slices ───────────────────────────────────────────────
    for slice_name in proposed_slices:
        if slice_name in spec_cache.slices:
            pass  # valid
        elif slice_name in spec_cache.segments:
            near_misses.append(NearMiss(name=slice_name, why_not="wrong_dimension_kind"))
        else:
            near_misses.append(NearMiss(name=slice_name, why_not="unknown_slice"))

    # ── 3. Validate segment ──────────────────────────────────────────────
    if proposed_segment is not None:
        if proposed_segment in spec_cache.segments:
            pass  # valid
        elif proposed_segment in spec_cache.slices:
            near_misses.append(NearMiss(name=proposed_segment, why_not="wrong_dimension_kind"))
        else:
            near_misses.append(NearMiss(name=proposed_segment, why_not="unknown_segment"))

    # ── 4. Validate by_entity ────────────────────────────────────────────
    if intent.by_entity is not None:
        entities = metric_spec.entities or []
        if intent.by_entity not in entities:
            near_misses.append(NearMiss(
                name=proposed_metric_name, why_not="unsupported_by_entity"
            ))

    # ── 5. Validate period_type ──────────────────────────────────────────
    if intent.period_type != "all_time" and not metric_spec.timestamp_col:
        near_misses.append(NearMiss(
            name=proposed_metric_name, why_not="unsupported_period_type"
        ))

    # ── Result ───────────────────────────────────────────────────────────
    if near_misses:
        return SpecMatchResult(exact_match=None, near_misses=near_misses)

    return SpecMatchResult(
        exact_match=ExactMatch(
            spec_token="",  # caller (resolve_intent tool) mints and fills this
            metric_name=proposed_metric_name,
            slices=proposed_slices,
            segment=proposed_segment,
        ),
        near_misses=[],
    )

DefinitionBot

DefinitionBot

aitaem.agent.definition_bot.DefinitionBot

Bases: Bot

Convenience bot for defining MetricSpec, SliceSpec, and SegmentSpec.

Uses a 4-step token-gated workflow: record_definition_intent → list_tables / describe_table → draft_spec → validate_spec loop → DefinitionOutput.

The bot is primarily a single-turn bot — use ask() for each new spec. chat() is provided for cross-turn context but multi-turn revision re-drafts from scratch (drafts are ephemeral per run(); see ND-10).

Construction

bot = DefinitionBot( model="anthropic:claude-sonnet-4-6", connection_manager=my_connection_manager, spec_cache=my_spec_cache, ) response = await bot.ask("Define a metric for weekly active users on the events table.")

Source code in aitaem/agent/definition_bot.py 
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
class DefinitionBot(Bot):
    """Convenience bot for defining MetricSpec, SliceSpec, and SegmentSpec.

    Uses a 4-step token-gated workflow: record_definition_intent → list_tables /
    describe_table → draft_spec → validate_spec loop → DefinitionOutput.

    The bot is primarily a single-turn bot — use ask() for each new spec.
    chat() is provided for cross-turn context but multi-turn revision re-drafts
    from scratch (drafts are ephemeral per run(); see ND-10).

    Construction:
        bot = DefinitionBot(
            model="anthropic:claude-sonnet-4-6",
            connection_manager=my_connection_manager,
            spec_cache=my_spec_cache,
        )
        response = await bot.ask("Define a metric for weekly active users on the events table.")
    """

    def __init__(
        self,
        *,
        model: Any,
        connection_manager: Any,
        spec_cache: Any,
        tenant_id: str | None = None,
        tools: list[Any] | None = None,
    ) -> None:
        # Set bot-specific resources BEFORE super().__init__() — _build_agent()
        # is called inside super().__init__() and needs these attributes.
        self._connection_manager = connection_manager
        self._spec_cache = spec_cache
        self._tenant_id = tenant_id
        super().__init__(model=model, tools=tools)
        self._conversation_id: str | None = None

    def _build_agent(self) -> Any:
        from pydantic_ai import Agent
        from pydantic_ai.toolsets import FunctionToolset
        from pydantic_ai.capabilities import ReinjectSystemPrompt

        toolset = FunctionToolset()
        toolset.add_function(record_definition_intent)  # Step 1
        toolset.add_function(list_tables)               # Step 2a
        toolset.add_function(describe_table)            # Step 2b
        toolset.add_function(draft_spec)                # Step 3
        toolset.add_function(validate_spec)             # Step 4

        for tool in self._tools:
            _register_tool(toolset, tool)
        self._toolset = toolset

        static_instructions = (
            _build_layer_a_definition()
            + "\n\n"
            + _build_layer_b_definition(self._spec_cache)
        )

        tenant_id = self._tenant_id or _definition_permission_fingerprint(self._spec_cache)

        agent = Agent(  # type: ignore[call-overload]
            model=self._model,
            deps_type=DefinitionDeps,
            output_type=DefinitionOutput,
            toolsets=[toolset],
            instructions=static_instructions,
            model_settings=_provider_cache_config_definition(self._model, tenant_id),
            capabilities=[ReinjectSystemPrompt(replace_existing=True)],
        )

        @agent.instructions
        def _layer_c() -> str:
            from datetime import date

            return (
                "# ─── Layer C: per-turn context ─────────────────────────────────────────────\n\n"
                f"Today is {date.today().isoformat()}."
            )

        return agent

    async def ask(
        self,
        message: str,
        *,
        extra_tools: list[Any] | None = None,
    ) -> DefinitionResponse:
        """Send a single-turn message. Does NOT accumulate history.

        Each ask() call is fully independent — drafts from prior ask() calls
        are not accessible. Use chat() for cross-turn context.
        """
        from datetime import datetime, timezone
        from aitaem.agent.trace import assemble_trace

        run_start = datetime.now(timezone.utc)
        deps = DefinitionDeps(
            connection_manager=self._connection_manager,
            spec_cache=self._spec_cache,
            store=self._store,
        )
        try:
            run_kwargs: dict[str, Any] = {"deps": deps}
            extra_toolset = _build_extra_toolset(extra_tools)
            if extra_toolset is not None:
                run_kwargs["toolsets"] = [extra_toolset]
            result = await self._agent.run(message, **run_kwargs)
            output = cast(DefinitionOutput, result.output)
            trace = assemble_trace(result, run_start)
            self._conversation_id = trace.conversation_id
            payload = DefinitionBot._assemble_payload(output, self._store)
            return DefinitionResponse(
                status=output.status,
                narrative=output.narrative,
                trace=trace,
                reason=output.reason,
                payload=payload,
            )
        except Exception as exc:
            return DefinitionBot._error_response(exc, run_start, self._conversation_id)

    async def chat(
        self,
        message: str,
        *,
        extra_tools: list[Any] | None = None,
    ) -> DefinitionResponse:
        """Send a message in multi-turn mode. Accumulates history on the bot.

        Drafts from prior turns are not accessible across runs — cross-turn
        revision re-drafts from scratch. For specs whose YAML exceeds 800 chars,
        cross-turn recovery may be lossy (ND-10).
        """
        from datetime import datetime, timezone
        from aitaem.agent.trace import assemble_trace

        run_start = datetime.now(timezone.utc)
        deps = DefinitionDeps(
            connection_manager=self._connection_manager,
            spec_cache=self._spec_cache,
            store=self._store,
        )
        try:
            run_kwargs: dict[str, Any] = {
                "message_history": self._message_history,
                "deps": deps,
            }
            extra_toolset = _build_extra_toolset(extra_tools)
            if extra_toolset is not None:
                run_kwargs["toolsets"] = [extra_toolset]
            result = await self._agent.run(message, **run_kwargs)
            self._message_history = result.all_messages()
            output = cast(DefinitionOutput, result.output)
            trace = assemble_trace(result, run_start)
            self._conversation_id = trace.conversation_id
            payload = DefinitionBot._assemble_payload(output, self._store)
            return DefinitionResponse(
                status=output.status,
                narrative=output.narrative,
                trace=trace,
                reason=output.reason,
                payload=payload,
            )
        except Exception as exc:
            return DefinitionBot._error_response(exc, run_start, self._conversation_id)

    @staticmethod
    def _assemble_payload(
        output: DefinitionOutput, store: ResultStore
    ) -> DefinitionPayload:
        """Assemble DefinitionPayload from the LLM's DefinitionOutput.

        When status is not ok or spec_draft_token is None, returns an empty payload.
        Otherwise retrieves the validated YAML from the store, re-parses it, and
        populates the appropriate spec field.
        """
        if output.status != Status.ok or output.spec_draft_token is None:
            return DefinitionPayload()

        try:
            entry = store.get_text(output.spec_draft_token)
        except Exception:
            return DefinitionPayload()

        yaml_string = entry.text
        metadata = entry.metadata
        spec_type = metadata.get("spec_type")
        spec_name = metadata.get("spec_name")

        import json as _json
        raw_rc = metadata.get("referenced_columns")
        referenced_columns: dict[str, list[str]] | None = None
        if raw_rc:
            try:
                referenced_columns = _json.loads(raw_rc)
            except Exception:
                pass

        raw_warnings = metadata.get("warnings")
        warnings: list[str] = []
        if raw_warnings:
            try:
                warnings = _json.loads(raw_warnings)
            except Exception:
                pass

        metric_spec = None
        slice_spec = None
        segment_spec = None

        try:
            if spec_type == "metric":
                metric_spec = _parse_yaml_to_spec("metric", yaml_string)
            elif spec_type == "slice":
                slice_spec = _parse_yaml_to_spec("slice", yaml_string)
            elif spec_type == "segment":
                segment_spec = _parse_yaml_to_spec("segment", yaml_string)
        except Exception:
            # Re-parse exceptions are swallowed — YAML was already validated.
            pass

        return DefinitionPayload(
            spec_type=spec_type,  # type: ignore[arg-type]
            spec_name=spec_name,
            yaml_string=yaml_string,
            spec_draft_token=output.spec_draft_token,
            validation_warnings=warnings,
            referenced_columns=referenced_columns,
            metric_spec=metric_spec,
            slice_spec=slice_spec,
            segment_spec=segment_spec,
        )

    @staticmethod
    def _error_response(
        exc: Exception, run_start: Any, conversation_id: str | None
    ) -> DefinitionResponse:
        """Build a status=error DefinitionResponse when _agent.run() raises."""
        import uuid
        from aitaem.agent.trace import RunTrace, Usage

        trace = RunTrace(
            run_id=str(uuid.uuid4()),
            conversation_id=conversation_id or str(uuid.uuid4()),
            timestamp=run_start,
            tool_calls=[],
            usage=Usage(),
            error=f"{type(exc).__name__}: {exc}",
        )
        return DefinitionResponse(
            status=Status.error,
            narrative="The request could not be completed due to an unexpected error.",
            trace=trace,
            reason=str(exc),
            payload=DefinitionPayload(),
        )

ask async 

ask(message: str, *, extra_tools: list[Any] | None = None) -> DefinitionResponse

Send a single-turn message. Does NOT accumulate history.

Each ask() call is fully independent — drafts from prior ask() calls are not accessible. Use chat() for cross-turn context.

Source code in aitaem/agent/definition_bot.py 
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
async def ask(
    self,
    message: str,
    *,
    extra_tools: list[Any] | None = None,
) -> DefinitionResponse:
    """Send a single-turn message. Does NOT accumulate history.

    Each ask() call is fully independent — drafts from prior ask() calls
    are not accessible. Use chat() for cross-turn context.
    """
    from datetime import datetime, timezone
    from aitaem.agent.trace import assemble_trace

    run_start = datetime.now(timezone.utc)
    deps = DefinitionDeps(
        connection_manager=self._connection_manager,
        spec_cache=self._spec_cache,
        store=self._store,
    )
    try:
        run_kwargs: dict[str, Any] = {"deps": deps}
        extra_toolset = _build_extra_toolset(extra_tools)
        if extra_toolset is not None:
            run_kwargs["toolsets"] = [extra_toolset]
        result = await self._agent.run(message, **run_kwargs)
        output = cast(DefinitionOutput, result.output)
        trace = assemble_trace(result, run_start)
        self._conversation_id = trace.conversation_id
        payload = DefinitionBot._assemble_payload(output, self._store)
        return DefinitionResponse(
            status=output.status,
            narrative=output.narrative,
            trace=trace,
            reason=output.reason,
            payload=payload,
        )
    except Exception as exc:
        return DefinitionBot._error_response(exc, run_start, self._conversation_id)

chat async 

chat(message: str, *, extra_tools: list[Any] | None = None) -> DefinitionResponse

Send a message in multi-turn mode. Accumulates history on the bot.

Drafts from prior turns are not accessible across runs — cross-turn revision re-drafts from scratch. For specs whose YAML exceeds 800 chars, cross-turn recovery may be lossy (ND-10).

Source code in aitaem/agent/definition_bot.py 
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
async def chat(
    self,
    message: str,
    *,
    extra_tools: list[Any] | None = None,
) -> DefinitionResponse:
    """Send a message in multi-turn mode. Accumulates history on the bot.

    Drafts from prior turns are not accessible across runs — cross-turn
    revision re-drafts from scratch. For specs whose YAML exceeds 800 chars,
    cross-turn recovery may be lossy (ND-10).
    """
    from datetime import datetime, timezone
    from aitaem.agent.trace import assemble_trace

    run_start = datetime.now(timezone.utc)
    deps = DefinitionDeps(
        connection_manager=self._connection_manager,
        spec_cache=self._spec_cache,
        store=self._store,
    )
    try:
        run_kwargs: dict[str, Any] = {
            "message_history": self._message_history,
            "deps": deps,
        }
        extra_toolset = _build_extra_toolset(extra_tools)
        if extra_toolset is not None:
            run_kwargs["toolsets"] = [extra_toolset]
        result = await self._agent.run(message, **run_kwargs)
        self._message_history = result.all_messages()
        output = cast(DefinitionOutput, result.output)
        trace = assemble_trace(result, run_start)
        self._conversation_id = trace.conversation_id
        payload = DefinitionBot._assemble_payload(output, self._store)
        return DefinitionResponse(
            status=output.status,
            narrative=output.narrative,
            trace=trace,
            reason=output.reason,
            payload=payload,
        )
    except Exception as exc:
        return DefinitionBot._error_response(exc, run_start, self._conversation_id)

DefinitionResponse

aitaem.agent.definition_bot.DefinitionResponse

Bases: BotResponse[DefinitionPayload]

Concrete response type for DefinitionBot — narrows BotResponse's generic payload.

Source code in aitaem/agent/definition_bot.py 
47
48
class DefinitionResponse(BotResponse[DefinitionPayload]):
    """Concrete response type for DefinitionBot — narrows BotResponse's generic payload."""

DefinitionPayload

aitaem.agent.definition_types.DefinitionPayload

Bases: BaseModel

Assembled bot payload returned in DefinitionResponse.payload.

Source code in aitaem/agent/definition_types.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
class DefinitionPayload(BaseModel):
    """Assembled bot payload returned in DefinitionResponse.payload."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    spec_type: Literal["metric", "slice", "segment"] | None = None
    spec_name: str | None = None
    yaml_string: str | None = None
    spec_draft_token: str | None = None
    validation_warnings: list[str] = []
    # Column names referenced in SQL, keyed by table/source field name.
    referenced_columns: dict[str, list[str]] | None = None
    # Exactly one of these is set based on spec_type; others are None.
    metric_spec: Any | None = None
    slice_spec: Any | None = None
    segment_spec: Any | None = None

DefinitionIntent

aitaem.agent.definition_types.DefinitionIntent dataclass

Records what the user wants to define. Set once per run by record_definition_intent.

Source code in aitaem/agent/definition_types.py 
22
23
24
25
26
27
28
29
30
31
32
@dataclass
class DefinitionIntent:
    """Records what the user wants to define. Set once per run by record_definition_intent."""

    spec_type: Literal["metric", "slice", "segment"]
    description: str
    existing_yaml: str | None = None
    is_update: bool = False
    # Parsed from existing_yaml once at intent-recording time; None if not an update
    # or if existing_yaml was malformed.
    original_name: str | None = None

SpecDraft

aitaem.agent.definition_types.SpecDraft dataclass

Server-side storage for a not-yet-validated YAML string. Keyed by draft_id.

Source code in aitaem/agent/definition_types.py 
35
36
37
38
39
40
41
@dataclass
class SpecDraft:
    """Server-side storage for a not-yet-validated YAML string. Keyed by draft_id."""

    draft_id: str
    spec_type: Literal["metric", "slice", "segment"]
    yaml_string: str

ColumnInfo

aitaem.agent.definition_types.ColumnInfo

Bases: BaseModel

One column in a table schema.

Source code in aitaem/agent/definition_types.py 
116
117
118
119
120
class ColumnInfo(BaseModel):
    """One column in a table schema."""

    name: str
    dtype: str

ListTablesResult

aitaem.agent.definition_types.ListTablesResult

Bases: BaseModel

Returned by list_tables. Both fields may be non-empty on partial success.

Source code in aitaem/agent/definition_types.py 
123
124
125
126
127
128
129
class ListTablesResult(BaseModel):
    """Returned by list_tables. Both fields may be non-empty on partial success."""

    # Keyed by backend_type; contains the list of table names for each successful backend.
    tables: dict[str, list[str]] = {}
    # Per-backend failure messages for backends that errored.
    errors: dict[str, str] = {}

DescribeTableResult

aitaem.agent.definition_types.DescribeTableResult

Bases: BaseModel

Returned by describe_table.

Source code in aitaem/agent/definition_types.py 
132
133
134
135
136
137
138
139
class DescribeTableResult(BaseModel):
    """Returned by describe_table."""

    table_name: str
    backend_type: str
    columns: list[ColumnInfo] = []
    # Set on unknown backend or table-not-found; columns will be empty.
    error: str | None = None

DraftSpecResult

aitaem.agent.definition_types.DraftSpecResult

Bases: BaseModel

Returned by draft_spec.

Source code in aitaem/agent/definition_types.py 
142
143
144
145
146
147
148
class DraftSpecResult(BaseModel):
    """Returned by draft_spec."""

    draft_id: str
    spec_type: Literal["metric", "slice", "segment"]
    # First 800 chars of the YAML; truncated for large specs (800-char limit per ND-10).
    yaml_preview: str

ValidateSpecResult

aitaem.agent.definition_types.ValidateSpecResult

Bases: BaseModel

Returned by validate_spec — the anti-hallucination gate.

Source code in aitaem/agent/definition_types.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
class ValidateSpecResult(BaseModel):
    """Returned by validate_spec — the anti-hallucination gate."""

    model_config = ConfigDict(extra="forbid")

    # Set only when all five checks pass. The canonical ResultStore pointer
    # (ToolResult protocol, 03-component-architecture.md §2). Excluded from
    # serialization — the LLM only ever sees spec_draft_token, derived below.
    result_id: str | None = Field(default=None, exclude=True)
    # Structural / SQL / name-conflict / composite cross-ref failures.
    errors: list[ValidationIssue] = []
    # Live-schema column mismatches (best-effort; never a hard blocker).
    column_errors: list[ValidationIssue] = []
    # Soft warnings (e.g. connection unavailable for column check).
    warnings: list[str] = []
    # Referenced column names by table, populated on success.
    referenced_columns: dict[str, list[str]] | None = None
    # Tool-level failure (e.g. draft_id not found). Distinct from errors/column_errors.
    error: str | None = None

    @computed_field  # type: ignore[prop-decorator]
    @property
    def spec_draft_token(self) -> str | None:
        """LLM-facing token. The LLM must copy this verbatim into DefinitionOutput.spec_draft_token."""
        return self.result_id or None

spec_draft_token property 

spec_draft_token: str | None

LLM-facing token. The LLM must copy this verbatim into DefinitionOutput.spec_draft_token.

ValidationIssue

aitaem.agent.definition_types.ValidationIssue

Bases: BaseModel

One validation failure from validate_spec.

Source code in aitaem/agent/definition_types.py 
151
152
153
154
155
156
class ValidationIssue(BaseModel):
    """One validation failure from validate_spec."""

    field: str
    message: str
    suggestion: str | None = None