Skip to content
Open
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@ node_modules/
/test_tmp/
.mcp.json
CLAUDE.local.md
AGENTS.override.md
.grok/*
.codex/*
.agents/*
!.agents/skills/
Expand All @@ -39,6 +41,4 @@ CLAUDE.local.md
/.cursor/
/.devcontainer/
.claude/plans/

# AICA mini-PR diagnostic scratch (see team.pydantic.work)
local-notes/
2 changes: 1 addition & 1 deletion docs/capabilities.md
Original file line number Diff line number Diff line change
Expand Up @@ -697,7 +697,7 @@ Capabilities can hook into five lifecycle points, each with up to four variants:
| [`before_node_run`][pydantic_ai.capabilities.AbstractCapability.before_node_run] | `(ctx: RunContext, *, node: AgentNode) -> AgentNode` | Observe or replace the node before execution |
| [`after_node_run`][pydantic_ai.capabilities.AbstractCapability.after_node_run] | `(ctx: RunContext, *, node: AgentNode, result: NodeResult) -> NodeResult` | Modify the result (next node or `End`) |
| [`wrap_node_run`][pydantic_ai.capabilities.AbstractCapability.wrap_node_run] | `(ctx: RunContext, *, node: AgentNode, handler: WrapNodeRunHandler) -> NodeResult` | Wrap each graph node execution |
| [`on_node_run_error`][pydantic_ai.capabilities.AbstractCapability.on_node_run_error] | `(ctx: RunContext, *, node: AgentNode, error: BaseException) -> NodeResult` | Handle node errors (see [error hooks](#error-hooks)) |
| [`on_node_run_error`][pydantic_ai.capabilities.AbstractCapability.on_node_run_error] | `(ctx: RunContext, *, node: AgentNode, error: Exception) -> NodeResult` | Handle node errors (see [error hooks](#error-hooks)) |

[`wrap_node_run`][pydantic_ai.capabilities.AbstractCapability.wrap_node_run] fires for every node in the [agent graph](agent.md#iterating-over-an-agents-graph) ([`UserPromptNode`][pydantic_ai.UserPromptNode], [`ModelRequestNode`][pydantic_ai.ModelRequestNode], [`CallToolsNode`][pydantic_ai.CallToolsNode]). Override this to observe node transitions, add per-step logging, or modify graph progression:

Expand Down
152 changes: 152 additions & 0 deletions pydantic_ai_slim/pydantic_ai/_display.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,152 @@
"""Agent banner display — logo, agent info, and next-steps guide.

Prints a branded banner to stderr the first time an agent runs. Subsequent calls
for the same agent instance are no-ops. Each unique agent instance gets its own
banner.
"""

from __future__ import annotations

import platform
import sys
from datetime import datetime, timezone
from typing import TYPE_CHECKING, Any

from . import __version__

if TYPE_CHECKING:
from .agent import Agent as _Agent

# Track which agent instances have already been displayed (process-wide).
_displayed_agent_ids: set[int] = set()

LOGO_PREAMBLE = r"""
pydantic-ai v{version} • Python {python_version}
"""

LOGO = r"""
/\
/ \
/ \
/ \
/________\
/ || \
/ || \
\_____||_____/
"""

LOGO_TEXT = r"""
[agent]
name: {name}
model: {model}
output_type: {output_type}
tools: {tools}
logfire: {logfire_status}


"""

# line by line join of Logo and text for easier maintenance
LOGO_PLUS_TEXT = '\n'.join(
f'{logo_line}{text_line}'
for logo_line, text_line in zip(
LOGO.splitlines(),
LOGO_TEXT.splitlines(),
)
)

NEXT_STEPS = """
─── Next steps ────────────────────────────────────
• Setup Logfire (Free): https://ai.pydantic.dev/logfire/
• Configure Logfire: logfire.configure(...)
• Instrument this agent: logfire.instrument_pydantic_ai()
• Instrument its calls: .instrument_httpx · .instrument_openai
• Export anywhere: OTLP → Logfire · etc

• Docs: https://ai.pydantic.dev/logfire/
• Turn off this banner: os.environ['PYDANTIC_AI_HIDE_BANNER'] = '1'
Comment thread
strawgate marked this conversation as resolved.

"""

BANNER = f'{LOGO_PREAMBLE}\n{LOGO_PLUS_TEXT}\n{NEXT_STEPS}'


def _get_model_name(agent: _Agent[Any, Any]) -> str:
"""Get a display-friendly model name from the agent."""
model = agent.model
if model is None:
return '(not set)'
if isinstance(model, str):
return model
# Model instance
try:
return model.model_id
except Exception:
return getattr(model, 'model_name', str(model))


def _get_output_type_name(agent: _Agent[Any, Any]) -> str:
"""Get a display-friendly output type name."""
try:
ot = agent.output_type
return getattr(ot, '__name__', str(ot))
except Exception:
return 'str'
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LOW — Fallback output_type: 'str' is misleading

The exception path returns the literal string 'str'. If agent.output_type ever raises (e.g. a property error on a user subclass), the banner tells the user the agent outputs a str, which is likely not true — it could be a structured Pydantic model, an int, a custom class, anything. The fallback should be neutral ('(unknown)', `'<?>') so the displayed value is never silently wrong. A real failure here should also be loud enough to investigate, not hidden behind a plausible-looking default.



def _get_tool_names(agent: _Agent[Any, Any]) -> str:
"""Get comma-separated tool names from the agent's function toolset."""
try:
tools = agent._function_toolset.tools # type: ignore[union-attr]
if tools:
return ', '.join(tools.keys())
return '(none)'
except Exception:
return '(none)'


def _get_logfire_status(agent: _Agent[Any, Any]) -> str:
"""Check whether Logfire instrumentation is configured."""
try:
settings = agent._resolve_instrumentation_settings()
return 'configured' if settings is not None else 'not configured'
except Exception:
return 'unknown'


def display_agent_banner(agent: _Agent[Any, Any]) -> None:
"""Display the agent banner to stderr, once per agent instance per process.

Args:
agent: The agent to display the banner for.
"""
import os

if os.environ.get('PYDANTIC_AI_HIDE_BANNER'):
return

agent_id = id(agent)
Comment thread
strawgate marked this conversation as resolved.
Outdated
if agent_id in _displayed_agent_ids:
return
_displayed_agent_ids.add(agent_id)

name = agent.name or '(unnamed)'
model = _get_model_name(agent)
output_type = _get_output_type_name(agent)
tools = _get_tool_names(agent)
logfire_status = _get_logfire_status(agent)
python_version = platform.python_version()
timestamp = datetime.now(tz=timezone.utc).strftime('%Y-%m-%d %H:%M:%S')
Comment thread
strawgate marked this conversation as resolved.
Outdated

banner = BANNER.format(
version=__version__,
python_version=python_version,
timestamp=timestamp,
name=name,
model=model,
output_type=output_type,
tools=tools,
logfire_status=logfire_status,
)

print(f'\n{banner}\n', file=sys.stderr)
23 changes: 23 additions & 0 deletions pydantic_ai_slim/pydantic_ai/agent/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -872,6 +872,24 @@ def event_stream_handler(self) -> EventStreamHandler[AgentDepsT] | None:
def __repr__(self) -> str:
return f'{type(self).__name__}(model={self.model!r}, name={self.name!r}, end_strategy={self.end_strategy!r}, model_settings={self.model_settings!r}, output_type={self.output_type!r})'

def display_banner(self) -> None:
Comment thread
strawgate marked this conversation as resolved.
"""Display a banner with logo, agent info, and next steps.

Prints to stderr once per agent instance. Subsequent calls are no-ops.
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LOW — Stale docstring says "once per agent instance"

The actual behavior is once per process (driven by _display._banner_displayed: bool). As written, "Subsequent calls are no-ops" is technically true after the first call on any agent, but "once per agent instance" is misleading — a second, brand-new Agent(...) instance will not see the banner either, because the flag is process-wide. The matching docstring in _display.py:117 ("once per process") is the accurate description; align this one with it.

Called automatically on the first `run()` / `run_sync()` / `iter()` call.

Example:
```python
from pydantic_ai import Agent

agent = Agent('openai:gpt-5.2', name='my_agent')
agent.display_banner()
```
"""
from .._display import display_agent_banner

display_agent_banner(self)

@overload
def iter(
self,
Expand Down Expand Up @@ -1036,6 +1054,11 @@ async def main():
if infer_name and self.name is None:
self._infer_name(inspect.currentframe())

# Display agent banner on first run (once per agent instance, process-wide)
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LOW — Stale comment: "once per agent instance, process-wide"

Same wording bug as the display_banner docstring above. The implementation is once per process (single boolean _banner_displayed), and the phrase is internally inconsistent. Update to something like # Display agent banner on first run (once per process).

from .._display import display_agent_banner

display_agent_banner(self)

# Tool retries cannot be overridden per run: `int` is treated as the output budget. An explicit
# `retries={'tools': ...}` is rejected so the value isn't silently dropped.
retry_overrides = _normalize_agent_retry_overrides(retries, int_means='output')
Expand Down
17 changes: 17 additions & 0 deletions pydantic_ai_slim/pydantic_ai/durable_exec/temporal/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -80,11 +80,28 @@ def _workflow_runner(runner: WorkflowRunner | None) -> WorkflowRunner:
'anyio',
'sniffio',
'httpcore',
# `certifi` is imported lazily by `httpx`/`ssl` when a client builds its TLS context. A
# model constructed inside the workflow (e.g. a `gateway/` model resolved via
# `infer_model`) creates its own HTTP client there, so without passing `certifi` through
# alongside the rest of the HTTP stack Temporal warns that it was "imported after initial
# workflow load" (a hard error under `filterwarnings=error`).
'certifi',
# `fastmcp` (and the `mcp` SDK it transitively imports) calls `Path.expanduser` at
# import time when resolving its config directory — restricted by the workflow
# sandbox. Safe to pass through: the call only happens once at module init.
'fastmcp',
'mcp',
# The `anthropic` SDK (>=0.99.0) calls `Path.home()` during client construction to
# resolve its credentials/profile config directory (`~/.config/anthropic`) — restricted
# by the workflow sandbox. This trips when a model is constructed inside the workflow,
# e.g. a `gateway/anthropic:` or `anthropic:` model resolved lazily via `infer_model`.
# Safe to pass through: a deterministic, read-only config lookup.
'anthropic',
# The `google-genai` SDK lazily imports `google.auth` submodules (e.g.
# `google.auth.aio.credentials`) while constructing its client, which Temporal flags as
# "imported after initial workflow load" when a `gateway/google-cloud:` (or `google-*:`)
# model is built inside the workflow.
'google.auth',
# Used by fastmcp via py-key-value-aio
'beartype',
# Imported inside `logfire._internal.json_encoder` when running `logfire.info` inside an activity with attributes to serialize
Expand Down
73 changes: 72 additions & 1 deletion tests/test_temporal.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,13 @@
from pydantic_ai.direct import model_request_stream
from pydantic_ai.exceptions import ApprovalRequired, CallDeferred, ModelRetry, UserError
from pydantic_ai.messages import UploadedFile
from pydantic_ai.models import Model, ModelRequestParameters, create_async_http_client, infer_model_profile
from pydantic_ai.models import (
Model,
ModelRequestParameters,
create_async_http_client,
infer_model,
infer_model_profile,
)
from pydantic_ai.models.function import AgentInfo, FunctionModel
from pydantic_ai.models.instrumented import InstrumentationSettings
from pydantic_ai.models.test import TestModel
Expand Down Expand Up @@ -1396,6 +1402,71 @@ async def test_mcptoolset_dynamic_toolset_in_workflow(allow_model_requests: None
assert 'pydantic' in output.lower() or 'agent' in output.lower()


# Regression test for the workflow-sandbox passthrough list (`_workflow_runner` in
# `durable_exec/temporal/__init__.py`). A `gateway/` model named by string is constructed lazily via
# `infer_model` *inside* the workflow, so the provider's SDK is imported and its client built under
# the `SandboxedWorkflowRunner`. Provider SDKs touch the filesystem/env at construction time, which
# the sandbox forbids unless the SDK module is passed through. Every other test builds its model at
# module scope (outside the sandbox), so this seam was previously uncovered. Construction-only (no
# model request) keeps it deterministic.
@workflow.defn
class ConstructModelInWorkflow:
@workflow.run
async def run(self, model_name: str) -> str:
# We assert only that construction succeeds — no request is made.
return type(infer_model(model_name)).__name__


@pytest.mark.parametrize(
('model_name', 'expected_model_class'),
[
# Only `gateway/` providers exercise the sandbox: they import their SDK lazily inside
# `gateway_provider()`, so the import and client construction run *inside* the workflow. Direct
# providers (e.g. `anthropic:`) import their SDK at module level, which rides Temporal's
# transitive passthrough of `pydantic_ai` and never trips — so they give no regression coverage.
#
# The reported regression: `gateway/anthropic:` in-workflow tripped the `anthropic` SDK's
# `Path.home()` access.
pytest.param('gateway/anthropic:claude-sonnet-4-6', 'AnthropicModel', id='gateway-anthropic'),
# Canary: OpenAI needs no passthrough today; turns red here (not in a user's workflow) if a
# future SDK release makes a restricted call (e.g. reads `~/...`) during construction.
pytest.param('gateway/openai-chat:gpt-5', 'OpenAIChatModel', id='gateway-openai'),
# Positive coverage of the `google.auth` (+`certifi`) passthrough: `google-genai` lazily
# imports `google.auth` during construction, which the sandbox flags without it.
pytest.param('gateway/google-cloud:gemini-2.5-pro', 'GoogleModel', id='gateway-google'),
],
)
async def test_model_construction_in_workflow_passes_sandbox(
model_name: str,
expected_model_class: str,
client: Client,
monkeypatch: pytest.MonkeyPatch,
):
# Dummy credentials suffice since no request is made. The gateway key must encode a region
# (`pylf_v<n>_<region>_...`) so the base URL can be inferred.
monkeypatch.setenv('PYDANTIC_AI_GATEWAY_API_KEY', 'pylf_v1_us_0123456789abcdef')

async with Worker(
client,
task_queue=TASK_QUEUE,
workflows=[ConstructModelInWorkflow],
# A sandbox violation surfaces as a workflow *task* failure, which Temporal retries forever
# by default — so a regression would hang rather than fail. Promote any in-workflow exception
# (e.g. `RestrictedWorkflowAccessError`) to a workflow failure so it surfaces immediately.
workflow_failure_exception_types=[Exception],
):
# Without the SDK passed through this fails with a `WorkflowFailureError`: under the suite's
# warnings-as-errors, Temporal's "imported after initial workflow load" becomes a hard error;
# in production the SDK's restricted `Path.home()`/env access raises `RestrictedWorkflowAccessError`.
result = await client.execute_workflow(
ConstructModelInWorkflow.run,
args=[model_name],
id=f'construct_model_{re.sub(r"[^a-zA-Z0-9]", "_", model_name)}',
task_queue=TASK_QUEUE,
)
assert result == expected_model_class


async def test_temporal_agent():
assert isinstance(complex_temporal_agent.model, TemporalModel)
assert complex_temporal_agent.model.wrapped == complex_agent.model
Expand Down
Loading