Tools

The ToolNode for wrapping external MCP/UTCP tools, its configuration models, OAuth support, connection and artifact-extraction presets, and the typed tool error hierarchy.

tools

ToolNode package exports.

POPULAR_MCP_SERVERS module-attribute

POPULAR_MCP_SERVERS = {'github': ExternalToolConfig(name='github', transport=TransportType.MCP, connection='npx -y @modelcontextprotocol/server-github', auth_type=AuthType.OAUTH2_USER, description='GitHub repositories, issues, pull requests'), 'filesystem': ExternalToolConfig(name='filesystem', transport=TransportType.MCP, connection='npx -y @modelcontextprotocol/server-filesystem /data', auth_type=AuthType.NONE, description='Read/write local filesystem'), 'postgres': ExternalToolConfig(name='postgres', transport=TransportType.MCP, connection='npx -y @modelcontextprotocol/server-postgres', env={'DATABASE_URL': '${DATABASE_URL}'}, auth_type=AuthType.NONE, description='Query PostgreSQL databases'), 'slack': ExternalToolConfig(name='slack', transport=TransportType.MCP, connection='npx -y @modelcontextprotocol/server-slack', auth_type=AuthType.OAUTH2_USER, description='Slack channels, messages, users'), 'google-drive': ExternalToolConfig(name='google-drive', transport=TransportType.MCP, connection='npx -y @anthropic/mcp-server-google-drive', auth_type=AuthType.OAUTH2_USER, description='Google Drive files and folders'), 'duckduckgo': ExternalToolConfig(name='duckduckgo', transport=TransportType.MCP, connection='npx -y duckduckgo-mcp-server', auth_type=AuthType.NONE, description='Web search and content retrieval via DuckDuckGo'), 'brave-search': ExternalToolConfig(name='brave-search', transport=TransportType.MCP, connection='npx -y @anthropic/mcp-server-brave-search', env={'BRAVE_API_KEY': '${BRAVE_API_KEY}'}, auth_type=AuthType.NONE, description='Web search via Brave Search API (requires BRAVE_API_KEY env var)'), 'redis': ExternalToolConfig(name='redis', transport=TransportType.MCP, connection='uvx --from redis-mcp-server@latest redis-mcp-server', env={'REDIS_HOST': '${REDIS_HOST}', 'REDIS_PORT': '${REDIS_PORT}'}, auth_type=AuthType.NONE, description='Redis database operations (requires uvx/Python)')}

ARTIFACT_PRESETS module-attribute

ARTIFACT_PRESETS: dict[str, ArtifactExtractionConfig] = {'tableau': TABLEAU_ARTIFACT_PRESET, 'github': GITHUB_ARTIFACT_PRESET, 'filesystem': FILESYSTEM_ARTIFACT_PRESET, 'google-drive': GOOGLE_DRIVE_ARTIFACT_PRESET}

FILESYSTEM_ARTIFACT_PRESET module-attribute

FILESYSTEM_ARTIFACT_PRESET = ArtifactExtractionConfig(max_inline_size=50000, auto_artifact_large_content=True, binary_detection=BinaryDetectionConfig(enabled=True, min_size_for_detection=500, require_magic_bytes=True), resources=ResourceHandlingConfig(enabled=True, auto_read_if_size_under_bytes=100000, inline_text_if_under_chars=50000, cache_reads_to_artifacts=True), tool_fields={'read_file': [ArtifactFieldConfig(field_path='content', content_type='binary', mime_type=None, summary_template="Read file '{path}' ({size} bytes). Artifact ID: {artifact_id}")]}, default_binary_summary='File content stored as artifact ({mime_type}, {size} bytes). Artifact ID: {artifact_id}', default_text_summary='Large text file stored as artifact ({size} chars). Artifact ID: {artifact_id}')

GITHUB_ARTIFACT_PRESET module-attribute

GITHUB_ARTIFACT_PRESET = ArtifactExtractionConfig(max_inline_size=10000, auto_artifact_large_content=True, binary_detection=BinaryDetectionConfig(enabled=True, min_size_for_detection=1000, require_magic_bytes=True), resources=ResourceHandlingConfig(enabled=True, auto_read_if_size_under_bytes=50000, inline_text_if_under_chars=10000, cache_reads_to_artifacts=True), tool_fields={'get_file_contents': [ArtifactFieldConfig(field_path='content', content_type='binary', mime_type=None, summary_template="Retrieved file '{path}' ({size} bytes). Artifact ID: {artifact_id}")], 'download_artifact': [ArtifactFieldConfig(field_path='data', content_type='binary', mime_type='application/zip', summary_template="Downloaded artifact '{name}' ({size} bytes). Artifact ID: {artifact_id}")], 'get_release_asset': [ArtifactFieldConfig(field_path='content', content_type='binary', mime_type=None, summary_template="Downloaded release asset '{name}' ({size} bytes). Artifact ID: {artifact_id}")]}, default_binary_summary='GitHub content stored as artifact ({mime_type}, {size} bytes). Artifact ID: {artifact_id}')

GOOGLE_DRIVE_ARTIFACT_PRESET module-attribute

GOOGLE_DRIVE_ARTIFACT_PRESET = ArtifactExtractionConfig(max_inline_size=10000, auto_artifact_large_content=True, binary_detection=BinaryDetectionConfig(enabled=True, min_size_for_detection=1000, require_magic_bytes=True), resources=ResourceHandlingConfig(enabled=True, auto_read_if_size_under_bytes=100000, inline_text_if_under_chars=10000, cache_reads_to_artifacts=True), tool_fields={'download_file': [ArtifactFieldConfig(field_path='content', content_type='binary', mime_type=None, summary_template="Downloaded '{name}' from Google Drive ({size} bytes). Artifact ID: {artifact_id}")], 'export_document': [ArtifactFieldConfig(field_path='content', content_type='pdf', mime_type='application/pdf', summary_template="Exported document '{name}' as PDF ({size} bytes). Artifact ID: {artifact_id}")]}, default_binary_summary='Google Drive content stored as artifact ({mime_type}, {size} bytes). Artifact ID: {artifact_id}')

TABLEAU_ARTIFACT_PRESET module-attribute

TABLEAU_ARTIFACT_PRESET = ArtifactExtractionConfig(max_inline_size=5000, auto_artifact_large_content=True, binary_detection=BinaryDetectionConfig(enabled=True, min_size_for_detection=500, require_magic_bytes=True), resources=ResourceHandlingConfig(enabled=True, auto_read_if_size_under_bytes=0, inline_text_if_under_chars=5000, cache_reads_to_artifacts=True), tool_fields={'download_workbook': [ArtifactFieldConfig(field_path='content', content_type='pdf', mime_type='application/pdf', summary_template="Downloaded workbook '{name}' as PDF ({size} bytes). Artifact ID: {artifact_id}")], 'get_view_as_pdf': [ArtifactFieldConfig(field_path='pdf_data', content_type='pdf', mime_type='application/pdf', summary_template="Exported view '{view_name}' as PDF ({size} bytes). Artifact ID: {artifact_id}")], 'get_view_as_image': [ArtifactFieldConfig(field_path='image_data', content_type='image', mime_type='image/png', summary_template="Exported view '{view_name}' as image ({size} bytes). Artifact ID: {artifact_id}")], 'export_dashboard': [ArtifactFieldConfig(field_path='content', content_type='pdf', mime_type='application/pdf', summary_template="Exported dashboard '{name}' ({size} bytes). Artifact ID: {artifact_id}")]}, default_binary_summary='Tableau content stored as artifact ({mime_type}, {size} bytes). Artifact ID: {artifact_id}')

ToolNode dataclass

ToolNode(config: ExternalToolConfig, registry: ModelRegistry, auth_manager: Any | None = None, _mcp_client: Any | None = None, _utcp_client: Any | None = None, _tools: list[NodeSpec] = list(), _tool_name_map: dict[str, str] = dict(), _connected: bool = False, _connected_loop: Any | None = None, _resource_cache: ResourceCache | None = None, _subscription_manager: ResourceSubscriptionManager | None = None, _resources: list[ResourceInfo] = list(), _resource_templates: list[ResourceTemplateInfo] = list(), _resources_supported: bool = False, _resource_update_callback: Callable[[str], None] | None = None, _prompts: list[PromptInfo] = list(), _prompts_supported: bool = False, _prompts_stale: bool = False, _app_metadata: dict[str, AppMetadata] = dict())

Unified external tool integration for Penguiflow (MCP only for Phase 1).

ToolNode connects to a single external tool source (an MCP server, or a UTCP/HTTP/CLI endpoint) and converts whatever tools/resources/prompts it exposes into Penguiflow NodeSpec objects that the planner and registry can consume. Callers typically construct one ToolNode per configured source, await connect() it, register get_tools() with the planner, invoke tools through call(), and await close() it during shutdown.

Attributes:

Name	Type	Description
config	ExternalToolConfig	Static configuration for the tool source (transport, auth, retry policy, artifact-extraction rules, etc.). See :class:~penguiflow.tools.config.ExternalToolConfig.
registry	ModelRegistry	The shared :class:~penguiflow.registry.ModelRegistry that discovered tool argument/result models are registered into.
auth_manager	Any | None	Optional OAuth manager used to resolve and refresh user tokens for AuthType.OAUTH2_USER sources (e.g. an :class:~penguiflow.tools.auth.OAuthManager). Required whenever config.auth_type is OAUTH2_USER; otherwise unused.

Example

node = ToolNode(config=cfg, registry=registry) await node.connect() specs = node.get_tools() await node.close()

config instance-attribute

config: ExternalToolConfig

registry instance-attribute

registry: ModelRegistry

auth_manager class-attribute instance-attribute

auth_manager: Any | None = None

resources_supported property

resources_supported: bool

Check if MCP resources are supported by this tool source.

resources property

resources: list[ResourceInfo]

Get cached list of resources (call list_resources for refresh).

resource_templates property

resource_templates: list[ResourceTemplateInfo]

Get cached list of resource templates.

app_tools property

app_tools: dict[str, AppMetadata]

Get all tools with MCP App metadata.

prompts_supported property

prompts_supported: bool

Check if MCP prompts are supported by this tool source.

prompts property

prompts: list[PromptInfo]

Get cached list of prompts (call list_prompts for refresh).

connect async

connect(ctx: ToolContext | None = None) -> None

Connect to tool source and discover available tools.

Idempotent: if already connected on the current event loop this is a no-op. If the connection was established on a different event loop (e.g. the ToolNode was built before an async server started), this transparently reconnects on the current loop. On success, discovered tools (and, for MCP transports, resources and prompts) are populated and registered into self.registry.

Parameters:

Name	Type	Description	Default
ctx	ToolContext | None	Optional ToolContext for HITL OAuth during connection. Required if auth_type is OAUTH2_USER.	None

Raises:

Type	Description
ToolAuthError	If auth_type is OAUTH2_USER and ctx is not provided, or the OAuth flow does not yield a token.
ToolConnectionError	If the transport is unsupported or the underlying MCP/UTCP client fails to connect.
ToolNodeError	If a discovered tool name collides with an already-registered name in self.registry.

get_tools

get_tools() -> list[NodeSpec]

Return discovered tools as Penguiflow NodeSpec entries.

Note: connect() must be called before this method.

Returns:

Type	Description
list[NodeSpec]	The list of :class:~penguiflow.catalog.NodeSpec discovered
list[NodeSpec]	during connect() (including any generated resource/prompt
list[NodeSpec]	tools). Empty if connect() has not been called yet.

get_tool_specs

get_tool_specs() -> list[NodeSpec]

Alias for get_tools for compatibility with generators.

Returns:

Type	Description
list[NodeSpec]	Same value as :meth:get_tools.

call async

call(tool_name: str, args: dict[str, Any], ctx: ToolContext) -> Any

Execute a tool with auth resolution and resilience.

Reconnects automatically if the client is not connected or was connected on a different event loop, resolves auth headers (pausing for HITL OAuth via ctx if needed), invokes the underlying MCP/UTCP tool with retries per config.retry_policy, and runs the result through the layered artifact-extraction pipeline (and, for MCP Apps-enabled tools, fetches and attaches the app UI resource).

Parameters:

Name	Type	Description	Default
tool_name	str	Namespaced tool name (e.g. "{source}.{tool}") or the bare original tool name; both are accepted.	required
args	dict[str, Any]	Tool call arguments as a plain dict (already validated against the tool's args model by the caller/planner).	required
ctx	ToolContext	ToolContext for the current run, used to resolve auth, pause for HITL flows, and access the artifact store.	required

Returns:

Type	Description
Any	A dict of the form {"result": <transformed_output>} where
Any	<transformed_output> has had binary/oversized content
Any	replaced by artifact references.

Raises:

Type	Description
ToolAuthError	If OAuth is required but cannot be resolved.
ToolNodeError	If no client is available or the call fails after exhausting retries.

close async

close() -> None

Clean up resources.

Marks the node disconnected, clears discovered tool state, and best -effort closes the underlying MCP/UTCP client (exceptions during client teardown, e.g. from a stale event loop, are swallowed). Safe to call multiple times.

list_resources async

list_resources(refresh: bool = False) -> list[ResourceInfo]

List available MCP resources.

Parameters:

Name	Type	Description	Default
refresh	bool	Force refresh from server	False

Returns:

Type	Description
list[ResourceInfo]	List of ResourceInfo objects (empty if not connected or resources not supported)

list_resource_templates async

list_resource_templates(refresh: bool = False) -> list[ResourceTemplateInfo]

List available MCP resource templates.

Parameters:

Name	Type	Description	Default
refresh	bool	Force refresh from server	False

Returns:

Type	Description
list[ResourceTemplateInfo]	List of ResourceTemplateInfo objects (empty if not connected or not supported)

read_resource async

read_resource(uri: str, ctx: ToolContext, *, use_cache: bool = True) -> dict[str, Any]

Read a resource by URI.

Uses ResourceCache to avoid repeated fetches. Binary content is stored in ArtifactStore; small text may be inlined.

Parameters:

Name	Type	Description	Default
uri	str	Resource URI to read	required
ctx	ToolContext	Tool context with artifact store access	required
use_cache	bool	Whether to use caching (default True)	True

Returns:

Type	Description
dict[str, Any]	Dict with 'artifact' (ArtifactRef) or 'text' (inline content) or 'error'

subscribe_resource async

subscribe_resource(uri: str, callback: Any | None = None) -> bool

Subscribe to resource updates.

Parameters:

Name	Type	Description	Default
uri	str	Resource URI to subscribe to	required
callback	Any | None	Optional callback for updates	None

Returns:

Type	Description
bool	True if subscription successful

unsubscribe_resource async

unsubscribe_resource(uri: str) -> bool

Unsubscribe from resource updates.

Parameters:

Name	Type	Description	Default
uri	str	Resource URI to unsubscribe from	required

Returns:

Type	Description
bool	True if unsubscription successful

handle_resource_updated

handle_resource_updated(uri: str) -> None

Handle a resource updated notification from MCP server.

Invalidates cache and notifies subscribers.

Parameters:

Name	Type	Description	Default
uri	str	URI of updated resource	required

set_resource_updated_callback

set_resource_updated_callback(callback: Callable[[str], None] | None) -> None

Register a callback invoked when MCP resources are updated.

Parameters:

Name	Type	Description	Default
callback	Callable[[str], None] | None	Callable invoked with the updated resource URI, or None to clear a previously registered callback.	required

has_app

has_app(tool_name: str) -> bool

Check if a tool has MCP App metadata.

Parameters:

Name	Type	Description	Default
tool_name	str	Namespaced tool name	required

Returns:

Type	Description
bool	True if the tool has app metadata

get_app_metadata

get_app_metadata(tool_name: str) -> AppMetadata | None

Get MCP App metadata for a tool.

Parameters:

Name	Type	Description	Default
tool_name	str	Namespaced tool name	required

Returns:

Type	Description
AppMetadata | None	AppMetadata or None

list_prompts async

list_prompts(refresh: bool = False) -> list[PromptInfo]

List available MCP prompts.

Parameters:

Name	Type	Description	Default
refresh	bool	Force refresh from server	False

Returns:

Type	Description
list[PromptInfo]	List of PromptInfo objects (empty if not connected or prompts not supported)

get_prompt async

get_prompt(name: str, arguments: dict[str, str] | None = None) -> dict[str, Any]

Execute a prompt and return rendered messages.

Parameters:

Name	Type	Description	Default
name	str	Prompt name	required
arguments	dict[str, str] | None	Optional arguments for the prompt	None

Returns:

Type	Description
dict[str, Any]	Dict with 'messages' (list of serialized messages) and 'description'

handle_prompts_changed

handle_prompts_changed() -> None

Handle a prompts/list_changed notification from MCP server.

Clears cached prompts so they are re-discovered on next access.

ExternalToolConfig

Bases: BaseModel

Configuration for a single external tool source consumed by ToolNode.

One instance describes how to reach and authenticate against one MCP server or UTCP/HTTP/CLI endpoint, plus the resilience and artifact-extraction policies applied to its tool calls. Validated at construction time by :meth:validate_config.

name class-attribute instance-attribute

name: str = Field(..., description="Unique namespace for tools (e.g., 'github')")

description class-attribute instance-attribute

description: str = Field(default='', description='Human-readable description of this tool source.')

transport instance-attribute

transport: TransportType

connection class-attribute instance-attribute

connection: str = Field(..., description='Connection string (command for MCP, URL for HTTP/UTCP)')

utcp_mode class-attribute instance-attribute

utcp_mode: UtcpMode = Field(default=UtcpMode.AUTO, description='For HTTP/UTCP: how to interpret connection (manual_url recommended)')

mcp_transport_mode class-attribute instance-attribute

mcp_transport_mode: McpTransportMode = Field(default=McpTransportMode.AUTO, description='For MCP over HTTP: auto-detect, sse, or streamable_http')

env class-attribute instance-attribute

env: dict[str, str] = Field(default_factory=dict, description='Environment variables passed to an MCP stdio subprocess, or used as UTCP variable substitutions. Values may reference ${VAR} to substitute from the host process environment.')

auth_type class-attribute instance-attribute

auth_type: AuthType = Field(default=AuthType.NONE, description='Authentication method used for this tool source.')

auth_config class-attribute instance-attribute

auth_config: dict[str, Any] = Field(default_factory=dict, description="Auth-type-specific settings, e.g. {'token': ...} for BEARER, {'api_key': ..., 'header': ...} for API_KEY, or {'cookie_name': ..., 'cookie_value': ...} for COOKIE. Values may reference ${VAR} for env substitution.")

timeout_s class-attribute instance-attribute

timeout_s: float = Field(default=30.0, ge=1.0, le=300.0, description='Per-call timeout in seconds.')

retry_policy class-attribute instance-attribute

retry_policy: RetryPolicy = Field(default_factory=RetryPolicy, description='Retry/backoff policy applied to tool calls.')

max_concurrency class-attribute instance-attribute

max_concurrency: int = Field(default=10, ge=1, le=100, description='Maximum number of concurrent in-flight tool calls for this source.')

tool_filter class-attribute instance-attribute

tool_filter: list[str] | None = Field(default=None, description='Regex patterns to include specific tools (None = all)')

arg_validation class-attribute instance-attribute

arg_validation: dict[str, Any] = Field(default_factory=lambda: {'emit_suspect': True}, description='Planner arg validation policy for discovered tools.')

artifact_extraction class-attribute instance-attribute

artifact_extraction: ArtifactExtractionConfig = Field(default_factory=ArtifactExtractionConfig, description='Configuration for extracting binary/large content as artifacts')

prompts class-attribute instance-attribute

prompts: PromptsConfig = Field(default_factory=lambda: PromptsConfig(), description='Configuration for MCP prompts discovery and tool generation')

apps class-attribute instance-attribute

apps: AppsConfig = Field(default_factory=lambda: AppsConfig(), description='Configuration for MCP Apps interactive UI support')

output_transformer class-attribute instance-attribute

output_transformer: OutputTransformer | None = Field(default=None, exclude=True, description='Custom async function to transform tool outputs')

validate_config

validate_config() -> ExternalToolConfig

Validate transport/auth-specific requirements after field parsing.

Raises:

Type	Description
ValueError	If auth_type requires keys in auth_config that are missing (e.g. BEARER needs token, API_KEY needs api_key, COOKIE needs cookie_name/cookie_value), or if utcp_mode is set to a non-default value while transport is MCP (utcp_mode only applies to HTTP/UTCP transports).

Returns:

Type	Description
ExternalToolConfig	The validated ExternalToolConfig instance (unchanged).

ArtifactExtractionConfig

Bases: BaseModel

Configuration for extracting artifacts from tool outputs.

max_inline_size class-attribute instance-attribute

max_inline_size: int = Field(default=10000, ge=0, description='Maximum chars before auto-artifact extraction')

auto_artifact_large_content class-attribute instance-attribute

auto_artifact_large_content: bool = Field(default=True, description='Automatically store large content as artifacts')

binary_detection class-attribute instance-attribute

binary_detection: BinaryDetectionConfig = Field(default_factory=BinaryDetectionConfig)

resources class-attribute instance-attribute

resources: ResourceHandlingConfig = Field(default_factory=ResourceHandlingConfig)

handle_mcp_typed_content class-attribute instance-attribute

handle_mcp_typed_content: bool = Field(default=True, description='Extract EmbeddedResource/blob typed content blocks')

tool_fields class-attribute instance-attribute

tool_fields: dict[str, list[ArtifactFieldConfig]] = Field(default_factory=dict, description='Map of tool_name -> list of field extraction configs')

default_binary_summary class-attribute instance-attribute

default_binary_summary: str = Field(default='Binary content stored as artifact ({mime_type}, {size} bytes). Artifact ID: {artifact_id}', description='Format string for the LLM-facing summary of extracted binary content. Supports {mime_type}, {size}, and {artifact_id} placeholders.')

default_text_summary class-attribute instance-attribute

default_text_summary: str = Field(default='Large text stored as artifact ({size} chars). Artifact ID: {artifact_id}', description='Format string for the LLM-facing summary of extracted oversized text. Supports {size} and {artifact_id} placeholders.')

ArtifactFieldConfig

Bases: BaseModel

Configuration for extracting specific fields as artifacts.

field_path class-attribute instance-attribute

field_path: str = Field(..., description="JSONPath or dot notation to the field (e.g., 'content' or 'result.pdf_data')")

content_type class-attribute instance-attribute

content_type: str = Field(..., description='Expected content type: pdf, image, binary, text')

mime_type class-attribute instance-attribute

mime_type: str | None = Field(default=None, description='Override MIME type (auto-detected if None)')

summary_template class-attribute instance-attribute

summary_template: str = Field(default='Downloaded {content_type} ({size} bytes)', description='Template for LLM summary')

AuthType

Bases: str, Enum

Authentication methods supported by ToolNode.

Values

NONE: No authentication. API_KEY: Static API key injected as a request header. BEARER: Static bearer token injected as an Authorization header. COOKIE: Cookie-based auth (e.g. a Databricks Apps session cookie). OAUTH2_USER: Per-user OAuth 2.0 authorization-code flow, resolved via a human-in-the-loop (HITL) pause using OAuthManager.

NONE class-attribute instance-attribute

NONE = 'none'

API_KEY class-attribute instance-attribute

API_KEY = 'api_key'

BEARER class-attribute instance-attribute

BEARER = 'bearer'

COOKIE class-attribute instance-attribute

COOKIE = 'cookie'

OAUTH2_USER class-attribute instance-attribute

OAUTH2_USER = 'oauth2_user'

BinaryDetectionConfig

Bases: BaseModel

Configuration for automatic binary content detection.

enabled class-attribute instance-attribute

enabled: bool = Field(default=True, description='Enable binary content detection')

signatures class-attribute instance-attribute

signatures: dict[str, tuple[str, str]] = Field(default_factory=lambda: dict(DEFAULT_BINARY_SIGNATURES), description='Map of base64 prefix -> (extension, mime_type)')

min_size_for_detection class-attribute instance-attribute

min_size_for_detection: int = Field(default=1000, ge=0, description='Minimum string length to check for binary content')

max_decode_bytes class-attribute instance-attribute

max_decode_bytes: int = Field(default=5000000, ge=1000, description='Maximum bytes to decode when probing binary content')

require_magic_bytes class-attribute instance-attribute

require_magic_bytes: bool = Field(default=True, description='Require magic byte validation after base64 decode')

ResourceHandlingConfig

Bases: BaseModel

Policy for MCP resources and resource_links.

enabled class-attribute instance-attribute

enabled: bool = Field(default=True, description='Enable resource link handling')

auto_read_if_size_under_bytes class-attribute instance-attribute

auto_read_if_size_under_bytes: int = Field(default=0, ge=0, description='Auto-read resources smaller than this (0 = never auto-read)')

inline_text_if_under_chars class-attribute instance-attribute

inline_text_if_under_chars: int = Field(default=10000, ge=0, description='Inline text resources smaller than this')

cache_reads_to_artifacts class-attribute instance-attribute

cache_reads_to_artifacts: bool = Field(default=True, description='Cache resource reads to artifact store')

RetryPolicy

Bases: BaseModel

Retry configuration for ToolNode tool calls, using tenacity semantics.

Consumed by ToolNode._call_with_retry, which retries only on retryable errors (timeouts, connection errors, or ToolNodeError subclasses marked retryable) with exponential backoff between attempts.

max_attempts class-attribute instance-attribute

max_attempts: int = Field(default=3, ge=1, le=10, description='Maximum number of call attempts, including the first (non-retry) attempt.')

wait_exponential_min_s class-attribute instance-attribute

wait_exponential_min_s: float = Field(default=0.1, ge=0.01, description='Minimum wait time in seconds before a retry (exponential backoff floor).')

wait_exponential_max_s class-attribute instance-attribute

wait_exponential_max_s: float = Field(default=5.0, ge=0.1, description='Maximum wait time in seconds before a retry (exponential backoff ceiling).')

retry_on_status class-attribute instance-attribute

retry_on_status: list[int] = Field(default_factory=lambda: [429, 500, 502, 503, 504], description='HTTP status codes that are treated as retryable when surfaced by adapters.')

TransportType

Bases: str, Enum

Supported communication protocols for an external tool source.

Values

MCP: Model Context Protocol via FastMCP; transport (stdio/SSE/HTTP) is auto-detected from ExternalToolConfig.connection. HTTP: REST API accessed via UTCP. UTCP: Native UTCP endpoint (manual or call-template based). CLI: Command-line tools invoked via UTCP's CLI call template.

MCP class-attribute instance-attribute

MCP = 'mcp'

HTTP class-attribute instance-attribute

HTTP = 'http'

UTCP class-attribute instance-attribute

UTCP = 'utcp'

CLI class-attribute instance-attribute

CLI = 'cli'

UtcpMode

Bases: str, Enum

How to interpret the UTCP connection string on ExternalToolConfig.

Values

AUTO: Try MANUAL_URL first (if the connection string looks like a manual document), falling back to BASE_URL otherwise. MANUAL_URL: connection points directly to a UTCP manual document (recommended: enables full tool discovery). BASE_URL: connection is a plain REST base URL with a synthesized call template (limited discovery, no manual document).

AUTO class-attribute instance-attribute

AUTO = 'auto'

MANUAL_URL class-attribute instance-attribute

MANUAL_URL = 'manual_url'

BASE_URL class-attribute instance-attribute

BASE_URL = 'base_url'

OAuthManager dataclass

OAuthManager(providers: dict[str, OAuthProviderConfig], token_store: TokenStore = InMemoryTokenStore(), _pending: dict[str, dict[str, float | str]] = dict())

Manages user OAuth flows with HITL integration.

Used by ToolNode when a tool source's auth_type is AuthType.OAUTH2_USER: the manager checks the token store for a cached token, and if none is available builds an authorization URL for the caller to present to the user (typically via a HITL pause), then completes the flow via :meth:handle_callback once the provider redirects back with an authorization code.

Attributes:

Name	Type	Description
providers	dict[str, OAuthProviderConfig]	Mapping of provider name to its :class:OAuthProviderConfig.
token_store	TokenStore	Backing :class:TokenStore used to cache and retrieve access tokens. Defaults to a fresh :class:InMemoryTokenStore.

providers instance-attribute

providers: dict[str, OAuthProviderConfig]

token_store class-attribute instance-attribute

token_store: TokenStore = field(default_factory=InMemoryTokenStore)

get_token async

get_token(user_id: str, provider: str) -> str | None

Look up a cached access token for a user/provider pair.

Parameters:

Name	Type	Description	Default
user_id	str	Identifier of the user to look up.	required
provider	str	OAuth provider name.	required

Returns:

Type	Description
str | None	The cached token, or None if none is stored or it expired.

get_auth_request

get_auth_request(provider: str, user_id: str, trace_id: str) -> dict[str, str | list[str]]

Build an authorization request for a HITL OAuth consent flow.

Generates and tracks a random state value (pruning any expired pending requests first) and constructs the provider's authorization URL. The returned payload is typically surfaced to the user via ctx.pause() so they can complete consent out-of-band.

Parameters:

Name	Type	Description	Default
provider	str	OAuth provider name (must exist in self.providers).	required
user_id	str	Identifier of the user initiating the flow.	required
trace_id	str	Trace ID of the run requesting authorization, used to resume the correct run in :meth:handle_callback.	required

Returns:

Type	Description
dict[str, str | list[str]]	Dict with display_name, auth_url (full authorization URL
dict[str, str | list[str]]	including the generated state), scopes, and state.

Raises:

Type	Description
ValueError	If provider is not a known provider name.

handle_callback async

handle_callback(code: str, state: str) -> tuple[str, str]

Complete an OAuth authorization-code exchange and cache the token.

Consumes the pending request created by :meth:get_auth_request, exchanges code for an access token at the provider's token endpoint, and stores the token in self.token_store for future :meth:get_token lookups.

Parameters:

Name	Type	Description	Default
code	str	Authorization code returned by the provider's redirect.	required
state	str	The state value returned alongside the code; must match a pending request created within the last 10 minutes.	required

Returns:

Type	Description
str	Tuple of (user_id, trace_id) identifying which user and run
str	initiated the flow, so the caller can resume the paused run.

Raises:

Type	Description
ValueError	If state is unknown/already consumed, the pending request has expired (older than 10 minutes), or the provider's token response contains an error.
RuntimeError	If aiohttp is not installed.

OAuthProviderConfig dataclass

OAuthProviderConfig(name: str, display_name: str, auth_url: str, token_url: str, client_id: str, client_secret: str, redirect_uri: str, scopes: list[str] = list())

Static configuration describing a single OAuth 2.0 authorization-code provider.

One instance is registered per provider in OAuthManager.providers, keyed by name.

Attributes:

Name	Type	Description
name	str	Unique provider key (matches the key used in OAuthManager.providers and the provider argument passed to :class:TokenStore methods).
display_name	str	Human-readable provider name shown to end users during the HITL consent flow.
auth_url	str	Provider's authorization endpoint (where the user is redirected to grant consent).
token_url	str	Provider's token endpoint used to exchange an authorization code for an access token.
client_id	str	OAuth client ID registered with the provider.
client_secret	str	OAuth client secret registered with the provider.
redirect_uri	str	Redirect URI registered with the provider that the authorization server will send the user back to.
scopes	list[str]	OAuth scopes to request during authorization. Defaults to an empty list (no scopes requested).

name instance-attribute

name: str

display_name instance-attribute

display_name: str

auth_url instance-attribute

auth_url: str

token_url instance-attribute

token_url: str

client_id instance-attribute

client_id: str

client_secret instance-attribute

client_secret: str

redirect_uri instance-attribute

redirect_uri: str

scopes class-attribute instance-attribute

scopes: list[str] = field(default_factory=list)

TokenStore

Bases: Protocol

Persistence contract for OAuth access tokens.

Implement this protocol to back :class:OAuthManager with a durable store (e.g. a database or secrets manager) instead of the default :class:InMemoryTokenStore. Tokens are keyed by (user_id, provider).

store async

store(user_id: str, provider: str, token: str, expires_at: float | None) -> None

Persist an access token for a user/provider pair.

Parameters:

Name	Type	Description	Default
user_id	str	Identifier of the user the token belongs to.	required
provider	str	OAuth provider name (matches :attr:OAuthProviderConfig.name).	required
token	str	The access token value to store.	required
expires_at	float | None	Unix timestamp (seconds) after which the token is considered expired, or None if it does not expire.	required

get async

get(user_id: str, provider: str) -> str | None

Retrieve a previously stored access token.

Parameters:

Name	Type	Description	Default
user_id	str	Identifier of the user the token belongs to.	required
provider	str	OAuth provider name.	required

Returns:

Type	Description
str | None	The stored token, or None if absent or expired.

delete async

delete(user_id: str, provider: str) -> None

Remove a stored access token, if present.

Parameters:

Name	Type	Description	Default
user_id	str	Identifier of the user the token belongs to.	required
provider	str	OAuth provider name.	required

InMemoryTokenStore

InMemoryTokenStore()

Simple in-memory :class:TokenStore for development and tests.

Tokens are held only in process memory (a plain dict keyed by (user_id, provider)) and are lost on process restart. Not suitable for production multi-worker deployments.

store async

store(user_id: str, provider: str, token: str, expires_at: float | None) -> None

Store a token in memory, overwriting any existing entry.

Parameters:

Name	Type	Description	Default
user_id	str	Identifier of the user the token belongs to.	required
provider	str	OAuth provider name.	required
token	str	The access token value to store.	required
expires_at	float | None	Unix timestamp (seconds) after which the token is considered expired, or None if it does not expire.	required

get async

get(user_id: str, provider: str) -> str | None

Fetch a token, transparently evicting it if expired.

Parameters:

Name	Type	Description	Default
user_id	str	Identifier of the user the token belongs to.	required
provider	str	OAuth provider name.	required

Returns:

Type	Description
str | None	The stored token, or None if absent or expired.

delete async

delete(user_id: str, provider: str) -> None

Remove a stored token, if present.

Parameters:

Name	Type	Description	Default
user_id	str	Identifier of the user the token belongs to.	required
provider	str	OAuth provider name.	required

ErrorCategory

Bases: str, Enum

Classification for retry decisions.

RETRYABLE_SERVER class-attribute instance-attribute

RETRYABLE_SERVER = 'retryable_server'

RETRYABLE_RATE_LIMIT class-attribute instance-attribute

RETRYABLE_RATE_LIMIT = 'retryable_rate'

NON_RETRYABLE_CLIENT class-attribute instance-attribute

NON_RETRYABLE_CLIENT = 'non_retryable'

AUTH_REQUIRED class-attribute instance-attribute

AUTH_REQUIRED = 'auth_required'

NETWORK class-attribute instance-attribute

NETWORK = 'network'

CANCELLED class-attribute instance-attribute

CANCELLED = 'cancelled'

UNKNOWN class-attribute instance-attribute

UNKNOWN = 'unknown'

ToolNodeError

ToolNodeError(message: str, *, status_code: int | None = None, category: ErrorCategory | None = None, cause: Exception | None = None)

Bases: Exception

Base exception for ToolNode errors.

category class-attribute instance-attribute

category: ErrorCategory = category or self._infer_category(status_code)

status_code class-attribute instance-attribute

status_code: int | None = status_code

is_retryable property

is_retryable: bool

retry_after_seconds property

retry_after_seconds: float | None

Hint for backoff, especially for 429s.

to_dict

to_dict() -> dict[str, str | int | bool | None]

ToolAuthError

ToolAuthError(message: str, *, status_code: int | None = None, category: ErrorCategory | None = None, cause: Exception | None = None)

Bases: ToolNodeError

Authentication required or failed.

category class-attribute instance-attribute

category = ErrorCategory.AUTH_REQUIRED

ToolClientError

ToolClientError(message: str, *, status_code: int | None = None, category: ErrorCategory | None = None, cause: Exception | None = None)

Bases: ToolNodeError

Client error (4xx) - don't retry.

category class-attribute instance-attribute

category = ErrorCategory.NON_RETRYABLE_CLIENT

ToolConnectionError

ToolConnectionError(message: str, *, status_code: int | None = None, category: ErrorCategory | None = None, cause: Exception | None = None)

Bases: ToolNodeError

Failed to connect to tool source.

category class-attribute instance-attribute

category = ErrorCategory.NETWORK

ToolRateLimitError

ToolRateLimitError(message: str, *, status_code: int | None = None, category: ErrorCategory | None = None, cause: Exception | None = None)

Bases: ToolNodeError

Rate limited by external service.

category class-attribute instance-attribute

category = ErrorCategory.RETRYABLE_RATE_LIMIT

ToolServerError

ToolServerError(message: str, *, status_code: int | None = None, category: ErrorCategory | None = None, cause: Exception | None = None)

Bases: ToolNodeError

Server error (5xx) - retry.

category class-attribute instance-attribute

category = ErrorCategory.RETRYABLE_SERVER

ToolTimeoutError

ToolTimeoutError(message: str, *, status_code: int | None = None, category: ErrorCategory | None = None, cause: Exception | None = None)

Bases: ToolNodeError

Tool execution exceeded timeout.

category class-attribute instance-attribute

category = ErrorCategory.RETRYABLE_SERVER

get_preset

get_preset(name: str) -> ExternalToolConfig

Get a pre-configured MCP server config by name.

Looks up name in :data:POPULAR_MCP_SERVERS. These presets default to npx -y ... stdio connections intended for local development; see the module docstring for production alternatives.

Parameters:

Name	Type	Description	Default
name	str	Preset key (e.g. "github", "filesystem", "slack").	required

Returns:

Type	Description
ExternalToolConfig	The preset's :class:~penguiflow.tools.config.ExternalToolConfig.
ExternalToolConfig	Note this returns the shared preset instance, not a copy.

Raises:

Type	Description
ValueError	If name is not a known preset.

get_artifact_preset

get_artifact_preset(name: str) -> ArtifactExtractionConfig

Get an artifact extraction preset by name.

Parameters:

Name	Type	Description	Default
name	str	Preset name (e.g., 'tableau', 'github', 'filesystem').	required

Returns:

Type	Description
ArtifactExtractionConfig	ArtifactExtractionConfig for the named preset.

Raises:

Type	Description
KeyError	If the preset name is not found.

Example

from penguiflow.tools.presets import get_artifact_preset

config = ExternalToolConfig( name="tableau", transport=TransportType.MCP, connection="http://tableau-mcp:8080/sse", artifact_extraction=get_artifact_preset("tableau"), )

get_artifact_preset_info

get_artifact_preset_info(name: str) -> dict[str, Any]

Get information about an artifact preset.

Parameters:

Name	Type	Description	Default
name	str	Preset name.	required

Returns:

Type	Description
dict[str, Any]	Dict with preset metadata.

get_artifact_preset_with_overrides

get_artifact_preset_with_overrides(name: str, **overrides: Any) -> ArtifactExtractionConfig

Get an artifact preset with field overrides.

Parameters:

Name	Type	Description	Default
name	str	Preset name.	required
**overrides	Any	Field overrides to apply.	{}

Returns:

Type	Description
ArtifactExtractionConfig	ArtifactExtractionConfig with overrides applied.

Example

config = get_artifact_preset_with_overrides( "tableau", max_inline_size=2000, )

list_artifact_presets

list_artifact_presets() -> list[str]

List available artifact preset names.

Returns:

Type	Description
list[str]	Sorted list of preset keys registered in :data:ARTIFACT_PRESETS
list[str]	(e.g. ["filesystem", "github", "google-drive", "tableau"]).

merge_artifact_preset

merge_artifact_preset(base: ArtifactExtractionConfig, preset_name: str) -> ArtifactExtractionConfig

Merge a preset's tool_fields into an existing config.

Useful for combining preset tool field mappings with custom settings.

Parameters:

Name	Type	Description	Default
base	ArtifactExtractionConfig	The base configuration to merge into.	required
preset_name	str	Name of the preset to merge from.	required

Returns:

Type	Description
ArtifactExtractionConfig	New ArtifactExtractionConfig with merged tool_fields.

Example

custom = ArtifactExtractionConfig(max_inline_size=2000) merged = merge_artifact_preset(custom, "tableau")

adapt_exception

adapt_exception(exc: Exception, transport: str) -> ToolNodeError

Route to appropriate adapter based on transport.

adapt_mcp_error

adapt_mcp_error(exc: Exception) -> ToolNodeError

Convert FastMCP exceptions to ToolNodeError.

adapt_utcp_error

adapt_utcp_error(exc: Exception) -> ToolNodeError

Convert UTCP exceptions to ToolNodeError.