From 31b3fd12178b5f7963a869e5ad349157382362c2 Mon Sep 17 00:00:00 2001 From: maplexu Date: Thu, 13 Aug 2026 15:31:10 -0400 Subject: [PATCH 1/5] AI-382: keep hosted tool credentials out of workflow history A hosted tool's credential is sent from workflow code to the model, so writing it into the tool config puts the credential in the workflow itself, on every model turn. There was no alternative: these are fields on OpenAI's own API types and the real token has to reach the provider. secret_reference() returns a placeholder carrying the name of an environment variable. The worker substitutes its value in _build_tool, immediately before the model call, so what the workflow holds is the variable's name. It applies to a hosted MCP tool's authorization and each header value, and to the value of each container domain secret on the hosted shell and code interpreter tools. Anything a provider or a remote MCP server sends back is deliberately out of scope. A counterparty that quotes a credential into its own error text is the counterparty's bug, and covering it would mean guessing at every way a string can be rendered. Also fixes a hosted ShellTool crash that blocked one of those sites: _build_tool passed an executor unconditionally, but upstream rejects one for a hosted environment, so every model turn raised UserError. --- temporalio/contrib/openai_agents/README.md | 30 + temporalio/contrib/openai_agents/__init__.py | 2 + .../openai_agents/_invoke_model_activity.py | 24 +- .../openai_agents/_secret_reference.py | 263 ++++++ .../openai_agents/test_openai_tool_secrets.py | 840 ++++++++++++++++++ 5 files changed, 1155 insertions(+), 4 deletions(-) create mode 100644 temporalio/contrib/openai_agents/_secret_reference.py create mode 100644 tests/contrib/openai_agents/test_openai_tool_secrets.py diff --git a/temporalio/contrib/openai_agents/README.md b/temporalio/contrib/openai_agents/README.md index 83539044c..2a0159fa8 100644 --- a/temporalio/contrib/openai_agents/README.md +++ b/temporalio/contrib/openai_agents/README.md @@ -487,6 +487,36 @@ A stateless factory that declares no parameters — like the `lambda: MCPServerS For network-accessible MCP servers, you can also use `HostedMCPTool` from the OpenAI Agents SDK, which uses an MCP client hosted by OpenAI. +## Secrets for Hosted Tools + +⚠️ **Experimental** - This functionality is subject to change prior to General Availability. + +Use `secret_reference()` for a hosted tool credential that should come from the worker's environment rather than being written into your workflow. Pass it the *name of an environment variable*, in place of the credential itself: + +```python +from agents import HostedMCPTool +from temporalio.contrib.openai_agents import secret_reference + +tool = HostedMCPTool( + tool_config={ + "type": "mcp", + "server_label": "my_server", + "server_url": "https://example.com/mcp", + "authorization": secret_reference("MY_MCP_TOKEN"), + } +) +``` + +Set `MY_MCP_TOKEN` on every worker that runs model activities. `secret_reference("MY_MCP_TOKEN")` returns the placeholder `temporal.secret_reference:MY_MCP_TOKEN`; each worker reads the variable from its own environment and sends that value on to the model provider in the placeholder's place. A worker without a value for it fails the model call with a non-retryable `ApplicationError` of type `SecretReferenceFailure`, naming the variable. + +The placeholder is substituted in these fields and no others: + +- `authorization`, and the value of each entry in `headers`, in a `HostedMCPTool`'s `tool_config` +- `value` in each entry of `network_policy.domain_secrets` under a hosted `ShellTool`'s `environment` +- `value` in each entry of `network_policy.domain_secrets` under a `CodeInterpreterTool`'s `container` + +Anywhere else — in a header *name*, or as an MCP server `factory_argument` (see [Factory Arguments](#factory-arguments)) — the placeholder is passed on as the literal string `temporal.secret_reference:MY_MCP_TOKEN`, and the receiving system gets text that is not a credential. Nothing in this SDK validates or complains about that; you find out from whatever that system does with it, typically a failed authentication. + ## Sandbox Support ⚠️ **Pre-release** - This functionality is subject to change prior to General Availability. diff --git a/temporalio/contrib/openai_agents/__init__.py b/temporalio/contrib/openai_agents/__init__.py index 3976f633c..3958612b7 100644 --- a/temporalio/contrib/openai_agents/__init__.py +++ b/temporalio/contrib/openai_agents/__init__.py @@ -9,6 +9,7 @@ StatelessMCPServerProvider, ) from temporalio.contrib.openai_agents._model_parameters import ModelActivityParameters +from temporalio.contrib.openai_agents._secret_reference import secret_reference from temporalio.contrib.openai_agents._temporal_openai_agents import ( OpenAIAgentsPlugin, OpenAIPayloadConverter, @@ -28,6 +29,7 @@ "SandboxClientProvider", "StatelessMCPServerProvider", "StatefulMCPServerProvider", + "secret_reference", "testing", "workflow", ] diff --git a/temporalio/contrib/openai_agents/_invoke_model_activity.py b/temporalio/contrib/openai_agents/_invoke_model_activity.py index 5435b6369..dd7559729 100644 --- a/temporalio/contrib/openai_agents/_invoke_model_activity.py +++ b/temporalio/contrib/openai_agents/_invoke_model_activity.py @@ -46,6 +46,11 @@ from temporalio import activity from temporalio.contrib.openai_agents._heartbeat_decorator import auto_heartbeater +from temporalio.contrib.openai_agents._secret_reference import ( + resolve_code_interpreter_tool_config, + resolve_mcp_tool_config, + resolve_shell_tool_environment, +) from temporalio.contrib.workflow_streams import WorkflowStreamClient from temporalio.exceptions import ApplicationError @@ -231,22 +236,33 @@ def _build_tool(tool: ToolInput) -> Tool: FileSearchTool, WebSearchTool, ImageGenerationTool, - CodeInterpreterTool, LocalShellTool, ToolSearchTool, ), ): return tool + elif isinstance(tool, CodeInterpreterTool): + return CodeInterpreterTool( + tool_config=resolve_code_interpreter_tool_config(tool.tool_config) + ) elif isinstance(tool, ShellToolInput): + environment = ( + None + if tool.environment is None + else resolve_shell_tool_environment(tool.environment) + ) + # Only a local environment takes an executor, and an absent type means + # local, matching how ShellTool normalizes its environment. + hosted = environment is not None and environment.get("type", "local") != "local" return ShellTool( name=tool.name, - environment=tool.environment, - executor=_noop_shell_executor, + environment=environment, + executor=None if hosted else _noop_shell_executor, ) elif isinstance(tool, ApplyPatchToolInput): return ApplyPatchTool(name=tool.name, editor=_NoopApplyPatchEditor()) elif isinstance(tool, HostedMCPToolInput): - return HostedMCPTool(tool_config=tool.tool_config) + return HostedMCPTool(tool_config=resolve_mcp_tool_config(tool.tool_config)) elif isinstance(tool, CustomToolInput): return CustomTool( name=tool.tool_config["name"], diff --git a/temporalio/contrib/openai_agents/_secret_reference.py b/temporalio/contrib/openai_agents/_secret_reference.py new file mode 100644 index 000000000..224109f15 --- /dev/null +++ b/temporalio/contrib/openai_agents/_secret_reference.py @@ -0,0 +1,263 @@ +"""References to secrets held in the worker process environment. + +A resolved secret is never written back into the activity's own input: the +resolving helpers copy shallowly at every level they write to, so the input goes +on holding the marker. +""" + +from __future__ import annotations + +import os +from collections.abc import Iterator, Mapping, MutableMapping +from typing import Any, cast + +from agents.tool import ShellToolContainerAutoEnvironment, ShellToolEnvironment +from openai.types.responses.tool_param import CodeInterpreter, Mcp +from pydantic import ValidationError + +from temporalio.contrib.openai_agents.workflow import AgentsWorkflowError +from temporalio.exceptions import ApplicationError + +_MARKER_PREFIX = "temporal.secret_reference:" + +_ERROR_TYPE = "SecretReferenceFailure" + + +def secret_reference(key: str) -> str: + """Refer to a secret held in the worker process environment. + + .. warning:: + This function is experimental and may change in future versions. + Use with caution in production environments. + + A hosted tool's credential is sent from workflow code to the model, so + writing the credential into the tool config puts the credential itself in + the workflow. Pass the *name of an environment variable* here instead, and + put the placeholder returned where the credential would have gone:: + + from agents import HostedMCPTool + + from temporalio.contrib.openai_agents import secret_reference + + tool = HostedMCPTool( + tool_config={ + "type": "mcp", + "server_label": "my_server", + "server_url": "https://example.com/mcp", + "authorization": secret_reference("MY_MCP_TOKEN"), + } + ) + + The worker reads the variable from its own environment and substitutes its + value for the placeholder immediately before the model call. + + Set the variable on every worker that runs model activities; a worker + without a value for it fails the model call with a non-retryable + ``ApplicationError`` of type ``SecretReferenceFailure``, naming the + variable. + + The placeholder returned is the string + ``"temporal.secret_reference:"``. It is substituted in a hosted MCP + tool's ``authorization`` and in the value of each of its ``headers``, and in + the ``value`` of each domain secret under a hosted shell or code interpreter + container's ``network_policy``. Anywhere else — in a header *name*, or as an + MCP server ``factory_argument`` — it reaches the receiving system as that + literal string. Nothing in this SDK validates or complains about one that + was not substituted; you find out from whatever that system does with text + that is not a credential, typically a failed authentication. + + Args: + key: Name of the environment variable to read on the worker. + + Returns: + A placeholder string to use in place of the secret. + + Raises: + AgentsWorkflowError: If ``key`` is empty. + """ + if not key: + raise AgentsWorkflowError( + "secret_reference() requires the name of an environment variable to read " + "on the worker, but the name given was empty." + ) + return _MARKER_PREFIX + key + + +def _resolve_secret_reference(value: str) -> str: + """Return ``value`` with a secret reference marker replaced by its secret. + + A string that is not a marker is returned unchanged. + + Raises: + ApplicationError: If the marker names no environment variable, or the + variable it names is unset or empty in the worker process + environment. Non-retryable, of type ``SecretReferenceFailure``. + """ + if not value.startswith(_MARKER_PREFIX): + return value + key = value[len(_MARKER_PREFIX) :] + if not key: + raise ApplicationError( + f"Malformed secret reference {value!r}: the text after " + f"{_MARKER_PREFIX!r} must be the name of an environment variable to read " + "on the worker. Build the placeholder with secret_reference().", + type=_ERROR_TYPE, + non_retryable=True, + ) + secret = os.environ.get(key) + if not secret: + raise ApplicationError( + f"Secret reference environment variable {key!r} is not set, or is empty, " + "in the worker process environment.", + type=_ERROR_TYPE, + non_retryable=True, + ) + return secret + + +def _shallow_copy(mapping: Any) -> Any: + """A writable plain ``dict`` copy of a mapping read off an activity argument. + + Anything resolved is written to the copy, never to the argument, which is + what leaves the activity's own input holding the marker. + """ + return dict(cast(Mapping[str, Any], mapping)) + + +def _malformed_domain_secret_error(e: ValidationError) -> ApplicationError: + """The rejection to raise for a domain secret that does not validate. + + pydantic rejects the whole entry for some malformed shapes and a single + field for others, so the type named is not claimed to be the entry's. + """ + error = e.errors()[0] + return ApplicationError( + f"Domain secret {error['loc'][0]} in a container network policy is " + f"malformed. Only its position and the type of the value that was " + f"rejected ({type(error['input']).__name__}) are reported: a malformed " + "entry could itself hold the secret.", + type=_ERROR_TYPE, + non_retryable=True, + ) + + +class _UnreadDomainSecrets: + """Stands in for domain secrets that a failed read left consumed. + + Every read of it raises that failure again, rather than coming back as a + policy with no domain secrets at all. + """ + + def __init__(self, error: ApplicationError) -> None: + """Hold the failure to raise.""" + self._error = error + + def __iter__(self) -> Iterator[Any]: + """Raise the failure that consumed the domain secrets.""" + raise self._error + + +def _resolve_network_policy(network_policy: Any) -> Any: + """Copy a container network policy, resolving each domain secret value. + + On the code interpreter path ``domain_secrets`` is declared as an iterable, + and pydantic deserializes it into a single-pass iterator. Reading it here + would leave the activity's own input holding nothing, so the entries — still + the markers the workflow sent — are materialized back onto the input before + the copy resolves them. + + Raises: + ApplicationError: If a marker cannot be resolved, or a domain secret is + malformed. Non-retryable. + """ + policy = cast(MutableMapping[str, Any], network_policy) + domain_secrets = policy.get("domain_secrets") + if domain_secrets is None: + return dict(policy) + try: + unresolved = list(domain_secrets) + except ValidationError as e: + error = _malformed_domain_secret_error(e) + # A failed read consumes the iterator too, so what goes back in its + # place fails the same way rather than reading as no secrets at all. + policy["domain_secrets"] = _UnreadDomainSecrets(error) + # Not chained: the validation error carries the entry it rejected. + raise error from None + policy["domain_secrets"] = unresolved + resolved = dict(policy) + resolved["domain_secrets"] = [ + _resolve_domain_secret(secret) for secret in unresolved + ] + return resolved + + +def _resolve_domain_secret(secret: Mapping[str, Any]) -> dict[str, Any]: + """Copy one domain secret, with its ``value`` resolved.""" + return {**secret, "value": _resolve_secret_reference(secret["value"])} + + +def resolve_mcp_tool_config(tool_config: Mcp) -> Mcp: + """Copy a hosted MCP tool config, resolving its authorization and headers. + + A header's name is passed through: a marker belongs where a credential + belongs, and a name is not that. + + Raises: + ApplicationError: If a marker cannot be resolved. Non-retryable. + """ + resolved = _shallow_copy(tool_config) + if "authorization" in tool_config: + resolved["authorization"] = _resolve_secret_reference( + tool_config["authorization"] + ) + headers = tool_config.get("headers") + if headers is not None: + resolved["headers"] = { + name: _resolve_secret_reference(value) for name, value in headers.items() + } + return resolved + + +def resolve_shell_tool_environment( + environment: ShellToolEnvironment, +) -> ShellToolEnvironment: + """Copy a shell tool environment, resolving its domain secret values. + + Only an auto-provisioned container has a network policy to carry secrets. + + Raises: + ApplicationError: If a marker cannot be resolved, or a domain secret is + malformed. Non-retryable. + """ + if environment.get("type") != "container_auto": + return _shallow_copy(environment) + auto = cast(ShellToolContainerAutoEnvironment, environment) + network_policy = auto.get("network_policy") + resolved = _shallow_copy(auto) + if network_policy is not None: + resolved["network_policy"] = _resolve_network_policy(network_policy) + return resolved + + +def resolve_code_interpreter_tool_config( + tool_config: CodeInterpreter, +) -> CodeInterpreter: + """Copy a code interpreter tool config, resolving its domain secret values. + + A container given by ID carries no network policy, so it has no secrets. + + Raises: + ApplicationError: If a marker cannot be resolved, or a domain secret is + malformed. Non-retryable. + """ + resolved = _shallow_copy(tool_config) + container = tool_config.get("container") + if not isinstance(container, Mapping): + return resolved + network_policy = container.get("network_policy") + if network_policy is None: + return resolved + resolved_container = _shallow_copy(container) + resolved_container["network_policy"] = _resolve_network_policy(network_policy) + resolved["container"] = resolved_container + return resolved diff --git a/tests/contrib/openai_agents/test_openai_tool_secrets.py b/tests/contrib/openai_agents/test_openai_tool_secrets.py new file mode 100644 index 000000000..7689c0d0e --- /dev/null +++ b/tests/contrib/openai_agents/test_openai_tool_secrets.py @@ -0,0 +1,840 @@ +"""Tests for secret references in hosted tool secrets.""" + +import uuid +from collections.abc import AsyncIterator +from typing import Any, cast + +import pytest +from agents import ( + AgentOutputSchemaBase, + CodeInterpreterTool, + Handoff, + HostedMCPTool, + Model, + ModelResponse, + ModelSettings, + ModelTracing, + Tool, + TResponseInputItem, + Usage, +) +from agents.items import TResponseStreamEvent +from agents.tool import ShellTool, ShellToolEnvironment + +from temporalio import workflow +from temporalio.api.failure.v1 import Failure +from temporalio.client import Client +from temporalio.contrib.openai_agents import ( + AgentsWorkflowError, + ModelActivityParameters, + OpenAIPayloadConverter, + secret_reference, +) +from temporalio.contrib.openai_agents._invoke_model_activity import ( + ActivityModelInput, + ModelActivity, + StreamingActivityModelInput, + _build_tool, +) +from temporalio.contrib.openai_agents._temporal_model_stub import _TemporalModelStub +from temporalio.contrib.openai_agents.testing import ( + AgentEnvironment, + TestModel, + TestModelProvider, +) +from temporalio.converter import DefaultFailureConverter, PayloadConverter +from temporalio.exceptions import ApplicationError +from temporalio.testing import ActivityEnvironment +from tests.helpers import new_worker + +# Fabricated secrets. Neither may reach a serialized activity argument. +SENTINEL = "sk-test-sentinel-4f1a9c7e2b" +ENV_KEY = "TEMPORAL_TEST_TOOL_SECRET" +OTHER_SENTINEL = "sk-test-other-8c3d5e0a1f" +OTHER_ENV_KEY = "TEMPORAL_TEST_OTHER_TOOL_SECRET" + + +def _round_trip_activity_input(tool: Tool) -> tuple[bytes, ActivityModelInput]: + """Serialize the activity arguments a workflow would send for ``tool``. + + Returns the serialized payload bytes and the input as the activity receives + it after deserialization. + """ + stub = _TemporalModelStub( + model_name="gpt-5", + model_params=ModelActivityParameters(), + agent=None, + ) + activity_input, _summary = stub._build_activity_input( + system_instructions=None, + input="hi", + model_settings=ModelSettings(), + tools=[tool], + output_schema=None, + handoffs=[], + tracing=ModelTracing.DISABLED, + previous_response_id=None, + conversation_id=None, + prompt=None, + ) + converter = OpenAIPayloadConverter() + payload = converter.to_payload(activity_input) + return payload.data, converter.from_payload(payload, ActivityModelInput) + + +def _activity_input_payload_and_tool(tool: Tool) -> tuple[bytes, Any]: + """The serialized payload, and the single tool the activity receives.""" + payload, received = _round_trip_activity_input(tool) + tools = received.get("tools") or [] + assert len(tools) == 1 + return payload, tools[0] + + +def _hosted_mcp_tool(authorization: str, header_value: str) -> HostedMCPTool: + return HostedMCPTool( + tool_config={ + "type": "mcp", + "server_label": "test_server", + "server_url": "https://example.com/mcp", + "authorization": authorization, + "headers": {"X-Token": header_value, "X-Plain": "not-a-secret"}, + } + ) + + +def _domain_secret(name: str, value: str) -> dict[str, str]: + return {"domain": "example.com", "name": name, "value": value} + + +def _allowlist(domain_secrets: tuple[Any, ...]) -> Any: + return { + "type": "allowlist", + "allowed_domains": ["example.com"], + "domain_secrets": list(domain_secrets), + } + + +def _shell_tool(*domain_secrets: Any) -> ShellTool: + environment: Any = { + "type": "container_auto", + "network_policy": _allowlist(domain_secrets), + } + return ShellTool(environment=environment) + + +def _code_interpreter_tool(*domain_secrets: Any) -> CodeInterpreterTool: + tool_config: Any = { + "type": "code_interpreter", + "container": { + "type": "auto", + "network_policy": _allowlist(domain_secrets), + }, + } + return CodeInterpreterTool(tool_config=tool_config) + + +def _fields(value: Any) -> dict[str, Any]: + """View a TypedDict-shaped value as a plain mapping, for assertions.""" + return cast(dict[str, Any], value) + + +def _domain_secrets(network_policy: Any) -> list[Any]: + return list(_fields(network_policy)["domain_secrets"]) + + +def _reported_failure(error: BaseException) -> Failure: + """The failure a worker would report to the server for ``error``. + + The converter walks ``__cause__``, or the implicit ``__context__`` when + there is none, into ``failure.cause``. + """ + failure = Failure() + DefaultFailureConverter().to_failure(error, PayloadConverter.default, failure) + return failure + + +def _shell_secrets(built: Tool) -> list[Any]: + assert isinstance(built, ShellTool) + assert built.environment is not None + return _domain_secrets(_fields(built.environment)["network_policy"]) + + +def _code_interpreter_secrets(built: Tool) -> list[Any]: + assert isinstance(built, CodeInterpreterTool) + container = _fields(built.tool_config)["container"] + return _domain_secrets(_fields(container)["network_policy"]) + + +def test_hosted_mcp_secrets_stay_out_of_activity_arguments( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv(ENV_KEY, SENTINEL) + marker = secret_reference(ENV_KEY) + + payload, received = _activity_input_payload_and_tool( + _hosted_mcp_tool(marker, marker) + ) + + assert SENTINEL.encode() not in payload + assert payload.count(marker.encode()) == 2 + assert received.tool_config["authorization"] == marker + assert received.tool_config["headers"]["X-Token"] == marker + + +def test_hosted_shell_domain_secret_stays_out_of_activity_arguments( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv(ENV_KEY, SENTINEL) + marker = secret_reference(ENV_KEY) + + payload, received = _activity_input_payload_and_tool( + _shell_tool(_domain_secret("TOKEN", marker)) + ) + + assert SENTINEL.encode() not in payload + assert marker.encode() in payload + secrets = _domain_secrets(received.environment["network_policy"]) + assert secrets[0]["value"] == marker + + +def test_code_interpreter_domain_secret_stays_out_of_activity_arguments( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv(ENV_KEY, SENTINEL) + marker = secret_reference(ENV_KEY) + + payload, received = _activity_input_payload_and_tool( + _code_interpreter_tool(_domain_secret("TOKEN", marker)) + ) + + assert SENTINEL.encode() not in payload + assert marker.encode() in payload + secrets = _domain_secrets(received.tool_config["container"]["network_policy"]) + assert secrets[0]["value"] == marker + + +def test_hosted_mcp_secrets_resolve_for_the_model_call( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv(ENV_KEY, SENTINEL) + marker = secret_reference(ENV_KEY) + _payload, received = _activity_input_payload_and_tool( + _hosted_mcp_tool(marker, marker) + ) + + built = _build_tool(received) + + assert isinstance(built, HostedMCPTool) + assert _fields(built.tool_config)["authorization"] == SENTINEL + assert _fields(built.tool_config)["headers"] == { + "X-Token": SENTINEL, + "X-Plain": "not-a-secret", + } + # The deserialized activity argument still holds the marker, so building + # again resolves the same secret rather than an emptied config. + assert received.tool_config["authorization"] == marker + assert received.tool_config["headers"]["X-Token"] == marker + + +def test_hosted_shell_domain_secret_resolves_for_the_model_call( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv(ENV_KEY, SENTINEL) + marker = secret_reference(ENV_KEY) + _payload, received = _activity_input_payload_and_tool( + _shell_tool(_domain_secret("TOKEN", marker)) + ) + + built = _build_tool(received) + + assert _shell_secrets(built) == [_domain_secret("TOKEN", SENTINEL)] + + +def test_code_interpreter_domain_secret_resolves_for_the_model_call( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv(ENV_KEY, SENTINEL) + marker = secret_reference(ENV_KEY) + _payload, received = _activity_input_payload_and_tool( + _code_interpreter_tool(_domain_secret("TOKEN", marker)) + ) + + built = _build_tool(received) + + assert _code_interpreter_secrets(built) == [_domain_secret("TOKEN", SENTINEL)] + + +def test_code_interpreter_domain_secret_survives_a_second_build( + monkeypatch: pytest.MonkeyPatch, +): + """``domain_secrets`` deserializes into a single-pass iterator here. + + Building twice from one deserialized input must resolve the secret both + times, and must leave the input itself holding the marker. + """ + monkeypatch.setenv(ENV_KEY, SENTINEL) + marker = secret_reference(ENV_KEY) + _payload, received = _activity_input_payload_and_tool( + _code_interpreter_tool(_domain_secret("TOKEN", marker)) + ) + + first = _build_tool(received) + second = _build_tool(received) + + assert _code_interpreter_secrets(first) == [_domain_secret("TOKEN", SENTINEL)] + assert _code_interpreter_secrets(second) == [_domain_secret("TOKEN", SENTINEL)] + assert _domain_secrets(received.tool_config["container"]["network_policy"]) == [ + _domain_secret("TOKEN", marker) + ] + + +def test_hosted_shell_domain_secret_survives_a_second_build( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv(ENV_KEY, SENTINEL) + marker = secret_reference(ENV_KEY) + _payload, received = _activity_input_payload_and_tool( + _shell_tool(_domain_secret("TOKEN", marker)) + ) + + first = _build_tool(received) + second = _build_tool(received) + + assert _shell_secrets(first) == [_domain_secret("TOKEN", SENTINEL)] + assert _shell_secrets(second) == [_domain_secret("TOKEN", SENTINEL)] + assert _domain_secrets(received.environment["network_policy"]) == [ + _domain_secret("TOKEN", marker) + ] + + +def test_only_marker_domain_secrets_are_resolved(monkeypatch: pytest.MonkeyPatch): + """A literal value alongside a marker is left exactly as the workflow sent it.""" + monkeypatch.setenv(ENV_KEY, SENTINEL) + marker = secret_reference(ENV_KEY) + literal = _domain_secret("PLAIN", "plain-token-value") + _payload, received = _activity_input_payload_and_tool( + _code_interpreter_tool(literal, _domain_secret("TOKEN", marker)) + ) + + built = _build_tool(received) + + assert _code_interpreter_secrets(built) == [ + literal, + _domain_secret("TOKEN", SENTINEL), + ] + + +def test_two_secret_references_in_one_mcp_config_resolve_to_their_own_secrets( + monkeypatch: pytest.MonkeyPatch, +): + """Each reference names its own variable, and stands for that one only.""" + monkeypatch.setenv(ENV_KEY, SENTINEL) + monkeypatch.setenv(OTHER_ENV_KEY, OTHER_SENTINEL) + _payload, received = _activity_input_payload_and_tool( + _hosted_mcp_tool(secret_reference(ENV_KEY), secret_reference(OTHER_ENV_KEY)) + ) + + built = _build_tool(received) + + assert isinstance(built, HostedMCPTool) + assert _fields(built.tool_config)["authorization"] == SENTINEL + assert _fields(built.tool_config)["headers"] == { + "X-Token": OTHER_SENTINEL, + "X-Plain": "not-a-secret", + } + + +def test_two_domain_secrets_resolve_to_their_own_secrets( + monkeypatch: pytest.MonkeyPatch, +): + """The entries are resolved one by one, each from the variable it names.""" + monkeypatch.setenv(ENV_KEY, SENTINEL) + monkeypatch.setenv(OTHER_ENV_KEY, OTHER_SENTINEL) + _payload, received = _activity_input_payload_and_tool( + _code_interpreter_tool( + _domain_secret("TOKEN", secret_reference(ENV_KEY)), + _domain_secret("OTHER", secret_reference(OTHER_ENV_KEY)), + ) + ) + + built = _build_tool(received) + + assert _code_interpreter_secrets(built) == [ + _domain_secret("TOKEN", SENTINEL), + _domain_secret("OTHER", OTHER_SENTINEL), + ] + + +def test_shell_domain_secrets_resolve_by_position(monkeypatch: pytest.MonkeyPatch): + """Each entry resolves from the variable it names, whichever position it holds.""" + monkeypatch.setenv(ENV_KEY, SENTINEL) + monkeypatch.setenv(OTHER_ENV_KEY, OTHER_SENTINEL) + literal = _domain_secret("PLAIN", "plain-token-value") + _payload, received = _activity_input_payload_and_tool( + _shell_tool( + literal, + _domain_secret("TOKEN", secret_reference(ENV_KEY)), + _domain_secret("OTHER", secret_reference(OTHER_ENV_KEY)), + ) + ) + + built = _build_tool(received) + + assert _shell_secrets(built) == [ + literal, + _domain_secret("TOKEN", SENTINEL), + _domain_secret("OTHER", OTHER_SENTINEL), + ] + + +def test_a_marker_in_a_header_name_is_passed_through( + monkeypatch: pytest.MonkeyPatch, +): + """A marker belongs where a credential belongs, and a header name is not that.""" + monkeypatch.setenv(ENV_KEY, SENTINEL) + marker = secret_reference(ENV_KEY) + tool_config: Any = { + "type": "mcp", + "server_label": "test_server", + "server_url": "https://example.com/mcp", + "headers": {marker: "not-a-secret"}, + } + + payload, received = _activity_input_payload_and_tool( + HostedMCPTool(tool_config=tool_config) + ) + + built = _build_tool(received) + + assert isinstance(built, HostedMCPTool) + assert _fields(built.tool_config)["headers"] == {marker: "not-a-secret"} + assert SENTINEL.encode() not in payload + + +def test_an_mcp_config_without_an_authorization_resolves_its_headers( + monkeypatch: pytest.MonkeyPatch, +): + """``authorization`` is optional, and an absent one is not one to resolve.""" + monkeypatch.setenv(ENV_KEY, SENTINEL) + tool_config: Any = { + "type": "mcp", + "server_label": "test_server", + "server_url": "https://example.com/mcp", + "headers": {"X-Token": secret_reference(ENV_KEY)}, + } + + _payload, received = _activity_input_payload_and_tool( + HostedMCPTool(tool_config=tool_config) + ) + + built = _build_tool(received) + + assert isinstance(built, HostedMCPTool) + assert "authorization" not in _fields(built.tool_config) + assert _fields(built.tool_config)["headers"] == {"X-Token": SENTINEL} + + +def test_an_mcp_config_without_headers_resolves_its_authorization( + monkeypatch: pytest.MonkeyPatch, +): + """``headers`` is optional, and an absent one is not one to resolve.""" + monkeypatch.setenv(ENV_KEY, SENTINEL) + tool_config: Any = { + "type": "mcp", + "server_label": "test_server", + "server_url": "https://example.com/mcp", + "authorization": secret_reference(ENV_KEY), + } + + _payload, received = _activity_input_payload_and_tool( + HostedMCPTool(tool_config=tool_config) + ) + + built = _build_tool(received) + + assert isinstance(built, HostedMCPTool) + assert _fields(built.tool_config)["authorization"] == SENTINEL + assert "headers" not in _fields(built.tool_config) + + +def test_local_shell_environment_keeps_its_executor(): + _payload, received = _activity_input_payload_and_tool( + ShellTool(environment={"type": "local"}, executor=lambda _request: "") + ) + + built = _build_tool(received) + + assert isinstance(built, ShellTool) + assert built.executor is not None + assert _fields(built.environment) == {"type": "local"} + + +@pytest.mark.parametrize( + "environment", + [ + {"type": "container_auto"}, + { + "type": "container_auto", + "network_policy": {"type": "disabled"}, + }, + {"type": "container_reference", "container_id": "cntr_abc"}, + ], + ids=["container_auto", "container_auto_disabled_policy", "container_reference"], +) +def test_hosted_shell_environment_gets_no_executor( + environment: ShellToolEnvironment, +): + """A hosted environment runs on OpenAI's side, and rejects an executor.""" + _payload, received = _activity_input_payload_and_tool( + ShellTool(environment=environment) + ) + + built = _build_tool(received) + + assert isinstance(built, ShellTool) + assert built.executor is None + assert _fields(built.environment) == environment + + +def test_code_interpreter_container_id_is_passed_through(): + """A container named by ID carries no network policy, so it has no secrets.""" + _payload, received = _activity_input_payload_and_tool( + CodeInterpreterTool( + tool_config={"type": "code_interpreter", "container": "cntr_abc"} + ) + ) + + built = _build_tool(received) + + assert isinstance(built, CodeInterpreterTool) + assert _fields(built.tool_config)["container"] == "cntr_abc" + + +@pytest.mark.parametrize( + "container", + [ + {"type": "auto"}, + {"type": "auto", "network_policy": {"type": "disabled"}}, + ], + ids=["no_policy", "disabled_policy"], +) +def test_code_interpreter_container_without_domain_secrets_is_passed_through( + container: Any, +): + """Only a policy carries domain secrets, and only an allowlist has any.""" + _payload, received = _activity_input_payload_and_tool( + CodeInterpreterTool( + tool_config={"type": "code_interpreter", "container": container} + ) + ) + + built = _build_tool(received) + + assert isinstance(built, CodeInterpreterTool) + assert _fields(built.tool_config)["container"] == container + + +@pytest.mark.parametrize("value", ["", None]) +def test_unset_or_empty_environment_variable_is_not_retryable( + monkeypatch: pytest.MonkeyPatch, value: str | None +): + if value is None: + monkeypatch.delenv(ENV_KEY, raising=False) + else: + monkeypatch.setenv(ENV_KEY, value) + marker = secret_reference(ENV_KEY) + _payload, received = _activity_input_payload_and_tool( + _hosted_mcp_tool(marker, marker) + ) + + with pytest.raises(ApplicationError) as err: + _build_tool(received) + + assert err.value.non_retryable + assert err.value.type == "SecretReferenceFailure" + assert ENV_KEY in err.value.message + + +def test_marker_without_a_variable_name_is_rejected_as_malformed(): + """The marker format is public, so a hand-written one can name nothing.""" + _payload, received = _activity_input_payload_and_tool( + _hosted_mcp_tool("temporal.secret_reference:", "plain") + ) + + with pytest.raises(ApplicationError) as err: + _build_tool(received) + + assert err.value.non_retryable + assert err.value.type == "SecretReferenceFailure" + assert "Malformed secret reference" in err.value.message + + +@pytest.mark.parametrize( + ("entry", "type_name"), + [ + ({"domain": "example.com", "name": "TOKEN", "vaule": SENTINEL}, "dict"), + ({"domain": "example.com", "name": "TOKEN", "value": 7}, "int"), + (None, "NoneType"), + (42, "int"), + (SENTINEL, "str"), + ], + ids=["mis_keyed", "wrong_value_type", "null", "number", "bare_string"], +) +def test_a_malformed_domain_secret_is_rejected_non_retryably( + entry: Any, type_name: str +): + """The rejection reports a position and a type, and nothing else. + + Neither the message nor the failure a worker reports may carry what + pydantic rejected, which for some shapes is the whole entry. + """ + _payload, received = _activity_input_payload_and_tool(_code_interpreter_tool(entry)) + + with pytest.raises(ApplicationError) as err: + _build_tool(received) + + assert err.value.non_retryable + assert err.value.type == "SecretReferenceFailure" + assert f"the type of the value that was rejected ({type_name})" in err.value.message + assert SENTINEL not in err.value.message + failure = _reported_failure(err.value) + assert not failure.HasField("cause") + assert SENTINEL.encode() not in failure.SerializeToString() + + +def test_a_malformed_domain_secret_is_reported_by_position( + monkeypatch: pytest.MonkeyPatch, +): + """A policy can carry several secrets, so the rejection says which one.""" + monkeypatch.setenv(ENV_KEY, SENTINEL) + _payload, received = _activity_input_payload_and_tool( + _code_interpreter_tool(_domain_secret("TOKEN", secret_reference(ENV_KEY)), 42) + ) + + with pytest.raises(ApplicationError) as err: + _build_tool(received) + + assert "Domain secret 1" in err.value.message + assert SENTINEL not in err.value.message + + +def test_a_malformed_domain_secret_is_rejected_again_on_a_second_build(): + """A failed read consumes the secrets, so a second build has to fail too.""" + _payload, received = _activity_input_payload_and_tool(_code_interpreter_tool(42)) + + with pytest.raises(ApplicationError): + _build_tool(received) + with pytest.raises(ApplicationError) as err: + _build_tool(received) + + assert err.value.non_retryable + assert err.value.type == "SecretReferenceFailure" + + +def test_a_policy_with_no_domain_secrets_is_passed_through(): + """No domain secrets is no secrets to resolve, and nothing to materialize.""" + policy: Any = {"type": "allowlist", "allowed_domains": ["example.com"]} + _payload, received = _activity_input_payload_and_tool( + CodeInterpreterTool( + tool_config={ + "type": "code_interpreter", + "container": {"type": "auto", "network_policy": policy}, + } + ) + ) + + built = _build_tool(received) + + assert isinstance(built, CodeInterpreterTool) + assert _fields(built.tool_config)["container"] == { + "type": "auto", + "network_policy": policy, + } + + +def test_plain_values_are_passed_through_unchanged( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv(ENV_KEY, SENTINEL) + plain = "temporal.secret_reference-but-not-quite" + _payload, received = _activity_input_payload_and_tool( + _hosted_mcp_tool(plain, plain) + ) + + built = _build_tool(received) + + assert isinstance(built, HostedMCPTool) + assert _fields(built.tool_config)["authorization"] == plain + assert _fields(built.tool_config)["headers"] == { + "X-Token": plain, + "X-Plain": "not-a-secret", + } + + +def test_secret_reference_rejects_an_empty_key(): + with pytest.raises(AgentsWorkflowError): + secret_reference("") + + +async def _no_stream_events() -> AsyncIterator[TResponseStreamEvent]: + """A stream that completes without events. + + Publishing one costs a real wait: the activity signals it to a workflow that + does not exist, and the flusher then retries for ten minutes. + """ + events: list[TResponseStreamEvent] = [] + for event in events: + yield event + + +class _ToolRecordingModel(Model): + """Records the tools the model activity hands the model.""" + + def __init__(self) -> None: + self.called = False + self.tools: list[Tool] = [] + + async def get_response( + self, + system_instructions: str | None, + input: str | list[TResponseInputItem], + model_settings: ModelSettings, + tools: list[Tool], + output_schema: AgentOutputSchemaBase | None, + handoffs: list[Handoff], + tracing: ModelTracing, + **kwargs: Any, + ) -> ModelResponse: + """Record the tools, and answer with nothing.""" + self.called = True + self.tools = tools + return ModelResponse(output=[], usage=Usage(), response_id=None) + + def stream_response( + self, + system_instructions: str | None, + input: str | list[TResponseInputItem], + model_settings: ModelSettings, + tools: list[Tool], + output_schema: AgentOutputSchemaBase | None, + handoffs: list[Handoff], + tracing: ModelTracing, + **kwargs: Any, + ) -> AsyncIterator[TResponseStreamEvent]: + """Record the tools, and stream nothing.""" + self.called = True + self.tools = tools + return _no_stream_events() + + +def _hosted_mcp_config_the_model_received(model: _ToolRecordingModel) -> dict[str, Any]: + """The tool config of the single hosted MCP tool the model was handed.""" + assert len(model.tools) == 1 + tool = model.tools[0] + assert isinstance(tool, HostedMCPTool) + return _fields(tool.tool_config) + + +async def test_invoke_model_activity_resolves_tool_secrets( + monkeypatch: pytest.MonkeyPatch, +): + """The model is handed the secret, not the marker the workflow sent.""" + monkeypatch.setenv(ENV_KEY, SENTINEL) + sent = _hosted_mcp_tool(secret_reference(ENV_KEY), "not-a-secret") + _payload, activity_input = _round_trip_activity_input(sent) + model = _ToolRecordingModel() + + await ActivityEnvironment().run( + ModelActivity(TestModelProvider(model)).invoke_model_activity, + activity_input, + ) + + assert _hosted_mcp_config_the_model_received(model) == { + **_fields(sent.tool_config), + "authorization": SENTINEL, + } + + +async def test_invoke_model_activity_rejects_a_malformed_domain_secret(): + """A rejection reaches the caller as the activity's own failure. + + A retryable escape would run the attempt again, with the same argument, + forever. + """ + _payload, activity_input = _round_trip_activity_input( + _code_interpreter_tool( + {"domain": "example.com", "name": "TOKEN", "vaule": SENTINEL} + ) + ) + model = _ToolRecordingModel() + + with pytest.raises(ApplicationError) as err: + await ActivityEnvironment().run( + ModelActivity(TestModelProvider(model)).invoke_model_activity, + activity_input, + ) + + assert err.value.non_retryable + assert err.value.type == "SecretReferenceFailure" + assert SENTINEL not in err.value.message + failure = _reported_failure(err.value) + assert not failure.HasField("cause") + assert SENTINEL.encode() not in failure.SerializeToString() + assert not model.called + + +async def test_invoke_model_activity_streaming_resolves_tool_secrets( + monkeypatch: pytest.MonkeyPatch, client: Client +): + """The streaming activity hands the model the secret by its own path.""" + monkeypatch.setenv(ENV_KEY, SENTINEL) + sent = _hosted_mcp_tool(secret_reference(ENV_KEY), "not-a-secret") + _payload, activity_input = _round_trip_activity_input(sent) + streaming_input: StreamingActivityModelInput = { + **activity_input, + "streaming_topic": "events", + } + model = _ToolRecordingModel() + + await ActivityEnvironment(client).run( + ModelActivity(TestModelProvider(model)).invoke_model_activity_streaming, + streaming_input, + ) + + assert _hosted_mcp_config_the_model_received(model) == { + **_fields(sent.tool_config), + "authorization": SENTINEL, + } + + +@workflow.defn +class SecretReferenceWorkflow: + """Builds a hosted MCP tool config the way a user's workflow would.""" + + @workflow.run + async def run(self, key: str) -> str: + tool_config: Any = { + "type": "mcp", + "server_label": "test_server", + "server_url": "https://example.com/mcp", + "authorization": secret_reference(key), + } + tool = HostedMCPTool(tool_config=tool_config) + return _fields(tool.tool_config)["authorization"] + + +async def test_secret_reference_can_be_called_from_workflow_code(client: Client): + """A marker built in workflow code reaches the workflow result unchanged.""" + async with AgentEnvironment( + model=TestModel.returning_responses([]), + ) as env: + client = env.applied_on_client(client) + async with new_worker(client, SecretReferenceWorkflow) as worker: + result = await client.execute_workflow( + SecretReferenceWorkflow.run, + ENV_KEY, + id=f"secret-reference-workflow-{uuid.uuid4()}", + task_queue=worker.task_queue, + ) + + assert result == f"temporal.secret_reference:{ENV_KEY}" From bcb2679960133b3ae9f4ba4c580c0414e93c5641 Mon Sep 17 00:00:00 2001 From: maplexu Date: Thu, 13 Aug 2026 17:41:20 -0400 Subject: [PATCH 2/5] AI-382: simplify the shell tool branch and cut private docstrings resolve_shell_tool_environment now takes the absent environment and returns the local one it stands for, which is what ShellTool would have normalized it to anyway. That leaves _build_tool with one conditional producing the executor, rather than a None ternary and a hosted flag feeding a second conditional. The private helpers in _secret_reference.py carried docstrings that narrated their own bodies. What is left names something a maintainer would otherwise break: the copy-at-every-level invariant that keeps a resolved secret out of the activity input, the single-pass iterator on the code interpreter path, and the reason the rejection is raised from None rather than chaining the pydantic error that holds the entry. --- .../openai_agents/_invoke_model_activity.py | 14 +++---- .../openai_agents/_secret_reference.py | 39 ++++--------------- 2 files changed, 15 insertions(+), 38 deletions(-) diff --git a/temporalio/contrib/openai_agents/_invoke_model_activity.py b/temporalio/contrib/openai_agents/_invoke_model_activity.py index dd7559729..bb5339a1a 100644 --- a/temporalio/contrib/openai_agents/_invoke_model_activity.py +++ b/temporalio/contrib/openai_agents/_invoke_model_activity.py @@ -225,6 +225,7 @@ async def _empty_on_invoke_handoff(_ctx: RunContextWrapper[Any], _input: str) -> async def _noop_shell_executor(*_a: Any, **_kw: Any) -> str: + """Satisfies the ShellExecutor type for tool reconstruction during model calls.""" return "" @@ -246,18 +247,17 @@ def _build_tool(tool: ToolInput) -> Tool: tool_config=resolve_code_interpreter_tool_config(tool.tool_config) ) elif isinstance(tool, ShellToolInput): - environment = ( - None - if tool.environment is None - else resolve_shell_tool_environment(tool.environment) - ) + environment = resolve_shell_tool_environment(tool.environment) # Only a local environment takes an executor, and an absent type means # local, matching how ShellTool normalizes its environment. - hosted = environment is not None and environment.get("type", "local") != "local" return ShellTool( name=tool.name, environment=environment, - executor=None if hosted else _noop_shell_executor, + executor=( + _noop_shell_executor + if environment.get("type", "local") == "local" + else None + ), ) elif isinstance(tool, ApplyPatchToolInput): return ApplyPatchTool(name=tool.name, editor=_NoopApplyPatchEditor()) diff --git a/temporalio/contrib/openai_agents/_secret_reference.py b/temporalio/contrib/openai_agents/_secret_reference.py index 224109f15..0bcd0e4e2 100644 --- a/temporalio/contrib/openai_agents/_secret_reference.py +++ b/temporalio/contrib/openai_agents/_secret_reference.py @@ -1,8 +1,7 @@ """References to secrets held in the worker process environment. A resolved secret is never written back into the activity's own input: the -resolving helpers copy shallowly at every level they write to, so the input goes -on holding the marker. +resolvers copy at every level they write to, so the input keeps the marker. """ from __future__ import annotations @@ -86,8 +85,6 @@ def secret_reference(key: str) -> str: def _resolve_secret_reference(value: str) -> str: """Return ``value`` with a secret reference marker replaced by its secret. - A string that is not a marker is returned unchanged. - Raises: ApplicationError: If the marker names no environment variable, or the variable it names is unset or empty in the worker process @@ -116,11 +113,6 @@ def _resolve_secret_reference(value: str) -> str: def _shallow_copy(mapping: Any) -> Any: - """A writable plain ``dict`` copy of a mapping read off an activity argument. - - Anything resolved is written to the copy, never to the argument, which is - what leaves the activity's own input holding the marker. - """ return dict(cast(Mapping[str, Any], mapping)) @@ -142,29 +134,20 @@ def _malformed_domain_secret_error(e: ValidationError) -> ApplicationError: class _UnreadDomainSecrets: - """Stands in for domain secrets that a failed read left consumed. - - Every read of it raises that failure again, rather than coming back as a - policy with no domain secrets at all. - """ + """Raises when iterated, so secrets a failed read consumed never read as absent.""" def __init__(self, error: ApplicationError) -> None: - """Hold the failure to raise.""" self._error = error def __iter__(self) -> Iterator[Any]: - """Raise the failure that consumed the domain secrets.""" raise self._error def _resolve_network_policy(network_policy: Any) -> Any: """Copy a container network policy, resolving each domain secret value. - On the code interpreter path ``domain_secrets`` is declared as an iterable, - and pydantic deserializes it into a single-pass iterator. Reading it here - would leave the activity's own input holding nothing, so the entries — still - the markers the workflow sent — are materialized back onto the input before - the copy resolves them. + On the code interpreter path pydantic deserializes ``domain_secrets`` into a + single-pass iterator, so the entries read here go back onto the input. Raises: ApplicationError: If a marker cannot be resolved, or a domain secret is @@ -178,8 +161,6 @@ def _resolve_network_policy(network_policy: Any) -> Any: unresolved = list(domain_secrets) except ValidationError as e: error = _malformed_domain_secret_error(e) - # A failed read consumes the iterator too, so what goes back in its - # place fails the same way rather than reading as no secrets at all. policy["domain_secrets"] = _UnreadDomainSecrets(error) # Not chained: the validation error carries the entry it rejected. raise error from None @@ -192,16 +173,12 @@ def _resolve_network_policy(network_policy: Any) -> Any: def _resolve_domain_secret(secret: Mapping[str, Any]) -> dict[str, Any]: - """Copy one domain secret, with its ``value`` resolved.""" return {**secret, "value": _resolve_secret_reference(secret["value"])} def resolve_mcp_tool_config(tool_config: Mcp) -> Mcp: """Copy a hosted MCP tool config, resolving its authorization and headers. - A header's name is passed through: a marker belongs where a credential - belongs, and a name is not that. - Raises: ApplicationError: If a marker cannot be resolved. Non-retryable. """ @@ -219,16 +196,18 @@ def resolve_mcp_tool_config(tool_config: Mcp) -> Mcp: def resolve_shell_tool_environment( - environment: ShellToolEnvironment, + environment: ShellToolEnvironment | None, ) -> ShellToolEnvironment: """Copy a shell tool environment, resolving its domain secret values. - Only an auto-provisioned container has a network policy to carry secrets. + An absent environment comes back as the local one ``ShellTool`` normalizes it to. Raises: ApplicationError: If a marker cannot be resolved, or a domain secret is malformed. Non-retryable. """ + if environment is None: + return {"type": "local"} if environment.get("type") != "container_auto": return _shallow_copy(environment) auto = cast(ShellToolContainerAutoEnvironment, environment) @@ -244,8 +223,6 @@ def resolve_code_interpreter_tool_config( ) -> CodeInterpreter: """Copy a code interpreter tool config, resolving its domain secret values. - A container given by ID carries no network policy, so it has no secrets. - Raises: ApplicationError: If a marker cannot be resolved, or a domain secret is malformed. Non-retryable. From bad62ac1f643d61a563a55951b197183d8afa0a6 Mon Sep 17 00:00:00 2001 From: maplexu Date: Fri, 14 Aug 2026 12:38:03 -0400 Subject: [PATCH 3/5] AI-382: cut the secret_reference docs to what a caller acts on The docstring restated its own contract, described what the worker does with the value, and named a placeholder format nobody types. The three substitution sites and the misplaced-placeholder caveat moved wholly to the README, where they read as a list with a working cross-reference, so each fact now lives on one surface instead of two. --- temporalio/contrib/openai_agents/README.md | 6 ++-- .../openai_agents/_secret_reference.py | 30 ++++++------------- 2 files changed, 12 insertions(+), 24 deletions(-) diff --git a/temporalio/contrib/openai_agents/README.md b/temporalio/contrib/openai_agents/README.md index 2a0159fa8..13ef5282f 100644 --- a/temporalio/contrib/openai_agents/README.md +++ b/temporalio/contrib/openai_agents/README.md @@ -507,15 +507,15 @@ tool = HostedMCPTool( ) ``` -Set `MY_MCP_TOKEN` on every worker that runs model activities. `secret_reference("MY_MCP_TOKEN")` returns the placeholder `temporal.secret_reference:MY_MCP_TOKEN`; each worker reads the variable from its own environment and sends that value on to the model provider in the placeholder's place. A worker without a value for it fails the model call with a non-retryable `ApplicationError` of type `SecretReferenceFailure`, naming the variable. +Set `MY_MCP_TOKEN` on every worker that runs model activities — if it is missing or empty there, the model call fails with a non-retryable error naming it. -The placeholder is substituted in these fields and no others: +The variable's value is substituted in these fields and no others: - `authorization`, and the value of each entry in `headers`, in a `HostedMCPTool`'s `tool_config` - `value` in each entry of `network_policy.domain_secrets` under a hosted `ShellTool`'s `environment` - `value` in each entry of `network_policy.domain_secrets` under a `CodeInterpreterTool`'s `container` -Anywhere else — in a header *name*, or as an MCP server `factory_argument` (see [Factory Arguments](#factory-arguments)) — the placeholder is passed on as the literal string `temporal.secret_reference:MY_MCP_TOKEN`, and the receiving system gets text that is not a credential. Nothing in this SDK validates or complains about that; you find out from whatever that system does with it, typically a failed authentication. +Anywhere else — a header *name*, or an MCP server `factory_argument` (see [Factory Arguments](#factory-arguments)) — the placeholder is passed on as literal text, with no error from this SDK. ## Sandbox Support diff --git a/temporalio/contrib/openai_agents/_secret_reference.py b/temporalio/contrib/openai_agents/_secret_reference.py index 0bcd0e4e2..cbf266047 100644 --- a/temporalio/contrib/openai_agents/_secret_reference.py +++ b/temporalio/contrib/openai_agents/_secret_reference.py @@ -29,10 +29,12 @@ def secret_reference(key: str) -> str: This function is experimental and may change in future versions. Use with caution in production environments. - A hosted tool's credential is sent from workflow code to the model, so - writing the credential into the tool config puts the credential itself in - the workflow. Pass the *name of an environment variable* here instead, and - put the placeholder returned where the credential would have gone:: + Use it for a hosted tool credential that should come from the worker's + environment rather than being written into your workflow. Put the + placeholder returned where the credential would have gone. Only the + variable's name is recorded in workflow history. + + :: from agents import HostedMCPTool @@ -47,23 +49,9 @@ def secret_reference(key: str) -> str: } ) - The worker reads the variable from its own environment and substitutes its - value for the placeholder immediately before the model call. - - Set the variable on every worker that runs model activities; a worker - without a value for it fails the model call with a non-retryable - ``ApplicationError`` of type ``SecretReferenceFailure``, naming the - variable. - - The placeholder returned is the string - ``"temporal.secret_reference:"``. It is substituted in a hosted MCP - tool's ``authorization`` and in the value of each of its ``headers``, and in - the ``value`` of each domain secret under a hosted shell or code interpreter - container's ``network_policy``. Anywhere else — in a header *name*, or as an - MCP server ``factory_argument`` — it reaches the receiving system as that - literal string. Nothing in this SDK validates or complains about one that - was not substituted; you find out from whatever that system does with text - that is not a credential, typically a failed authentication. + Set the variable on every worker that runs model activities — if it is + missing or empty there, the model call fails with a non-retryable + ``ApplicationError`` of type ``SecretReferenceFailure`` naming it. Args: key: Name of the environment variable to read on the worker. From 5c2a8cd03bd77ec8c5a82238296b477c01cd19ae Mon Sep 17 00:00:00 2001 From: maplexu Date: Mon, 17 Aug 2026 12:23:11 -0400 Subject: [PATCH 4/5] AI-382: bound worker env references to the names a plugin lists secret_reference() becomes temporal_worker_env_ref(), and the reference it returns is now delimited at both ends, temporal.worker_env_ref:{NAME}, so it can be substituted inside a larger value. "Bearer " plus a reference works, which a workflow could not otherwise express, since composing that string itself would mean holding the credential. A truncated or half-written reference matches nothing and travels on as ordinary text. A worker resolves only the names its plugin lists in resolvable_worker_env_vars, which is empty by default. The list exists so a tool config field built from text the workflow did not author cannot name any variable in the worker's environment and have it read. Each occurrence is checked on its own name before any environment access, so one value can hold a resolved credential beside the literal text of a name that was not listed. A literal "*" anywhere in the list gives up that bound and allows every name. AgentEnvironment forwards the same parameter, so a user can test their own usage through it. Failures stay quiet. A name the worker allows but has not set resolves to an empty value, because an empty value may be what the author meant, and crashing a worker over a value the workflow chose is worse. The malformed-domain-secret rejection is gone for the same reason: a typo that lets a credential escape is the author's to fix rather than ours to police. What does raise is a bare string passed as resolvable_worker_env_vars, since frozenset() would decompose it into characters and a value containing "*" would silently make every variable readable. The tool support table gains a ShellTool row, now that a hosted one reaches the model at all. --- temporalio/contrib/openai_agents/README.md | 19 +- temporalio/contrib/openai_agents/__init__.py | 6 +- .../openai_agents/_invoke_model_activity.py | 41 +- .../openai_agents/_secret_reference.py | 228 ------ .../openai_agents/_temporal_openai_agents.py | 13 +- .../openai_agents/_temporal_worker_env_ref.py | 153 ++++ temporalio/contrib/openai_agents/testing.py | 9 +- tests/contrib/openai_agents/test_openai.py | 7 +- .../openai_agents/test_openai_tool_secrets.py | 703 +++++++++--------- 9 files changed, 581 insertions(+), 598 deletions(-) delete mode 100644 temporalio/contrib/openai_agents/_secret_reference.py create mode 100644 temporalio/contrib/openai_agents/_temporal_worker_env_ref.py diff --git a/temporalio/contrib/openai_agents/README.md b/temporalio/contrib/openai_agents/README.md index 13ef5282f..f2a748100 100644 --- a/temporalio/contrib/openai_agents/README.md +++ b/temporalio/contrib/openai_agents/README.md @@ -491,23 +491,31 @@ For network-accessible MCP servers, you can also use `HostedMCPTool` from the Op ⚠️ **Experimental** - This functionality is subject to change prior to General Availability. -Use `secret_reference()` for a hosted tool credential that should come from the worker's environment rather than being written into your workflow. Pass it the *name of an environment variable*, in place of the credential itself: +Use `temporal_worker_env_ref()` for a hosted tool credential that should come from the worker's environment rather than being written into your workflow. Pass it the *name of an environment variable*, in place of the credential itself: ```python from agents import HostedMCPTool -from temporalio.contrib.openai_agents import secret_reference +from temporalio.contrib.openai_agents import temporal_worker_env_ref tool = HostedMCPTool( tool_config={ "type": "mcp", "server_label": "my_server", "server_url": "https://example.com/mcp", - "authorization": secret_reference("MY_MCP_TOKEN"), + "authorization": temporal_worker_env_ref("MY_MCP_TOKEN"), } ) ``` -Set `MY_MCP_TOKEN` on every worker that runs model activities — if it is missing or empty there, the model call fails with a non-retryable error naming it. +Every worker that runs model activities must both set `MY_MCP_TOKEN` and name it as resolvable: + +```python +plugin = OpenAIAgentsPlugin(resolvable_worker_env_vars=["MY_MCP_TOKEN"]) +``` + +Nothing resolves unless the worker names it. A reference to a name the worker does not allow is sent to the model provider as literal text. Names are matched exactly — no globbing, no prefixes — and `"*"` anywhere in the list allows every environment variable on the worker. + +A reference need not be the whole value: `"Bearer " + temporal_worker_env_ref("MY_MCP_TOKEN")` substitutes the credential inside a larger string. Each reference in a value is checked against the list on its own, so one value can end up holding both a resolved credential and the literal text of a name the worker does not allow. A name the worker does allow but has not set resolves to an empty value. The variable's value is substituted in these fields and no others: @@ -515,7 +523,7 @@ The variable's value is substituted in these fields and no others: - `value` in each entry of `network_policy.domain_secrets` under a hosted `ShellTool`'s `environment` - `value` in each entry of `network_policy.domain_secrets` under a `CodeInterpreterTool`'s `container` -Anywhere else — a header *name*, or an MCP server `factory_argument` (see [Factory Arguments](#factory-arguments)) — the placeholder is passed on as literal text, with no error from this SDK. +Anywhere else — in a header *name*, or as an MCP server `factory_argument` (see [Factory Arguments](#factory-arguments)) — the reference is passed on as literal text, with no error from this SDK. ## Sandbox Support @@ -749,6 +757,7 @@ Certain tools are not suitable for a distributed computing environment, so these | HostedMCPTool | Yes | | ImageGenerationTool | Yes | | CodeInterpreterTool | Yes | +| ShellTool | Yes | | ComputerTool | No | #### Tool Context diff --git a/temporalio/contrib/openai_agents/__init__.py b/temporalio/contrib/openai_agents/__init__.py index 3958612b7..3305eaf95 100644 --- a/temporalio/contrib/openai_agents/__init__.py +++ b/temporalio/contrib/openai_agents/__init__.py @@ -9,11 +9,13 @@ StatelessMCPServerProvider, ) from temporalio.contrib.openai_agents._model_parameters import ModelActivityParameters -from temporalio.contrib.openai_agents._secret_reference import secret_reference from temporalio.contrib.openai_agents._temporal_openai_agents import ( OpenAIAgentsPlugin, OpenAIPayloadConverter, ) +from temporalio.contrib.openai_agents._temporal_worker_env_ref import ( + temporal_worker_env_ref, +) from temporalio.contrib.openai_agents.sandbox._sandbox_client_provider import ( SandboxClientProvider, ) @@ -29,7 +31,7 @@ "SandboxClientProvider", "StatelessMCPServerProvider", "StatefulMCPServerProvider", - "secret_reference", + "temporal_worker_env_ref", "testing", "workflow", ] diff --git a/temporalio/contrib/openai_agents/_invoke_model_activity.py b/temporalio/contrib/openai_agents/_invoke_model_activity.py index bb5339a1a..b94d48beb 100644 --- a/temporalio/contrib/openai_agents/_invoke_model_activity.py +++ b/temporalio/contrib/openai_agents/_invoke_model_activity.py @@ -4,6 +4,7 @@ """ import enum +from collections.abc import Collection from dataclasses import dataclass from datetime import timedelta from typing import Any, NoReturn @@ -46,10 +47,8 @@ from temporalio import activity from temporalio.contrib.openai_agents._heartbeat_decorator import auto_heartbeater -from temporalio.contrib.openai_agents._secret_reference import ( - resolve_code_interpreter_tool_config, - resolve_mcp_tool_config, - resolve_shell_tool_environment, +from temporalio.contrib.openai_agents._temporal_worker_env_ref import ( + _WorkerEnvRefResolver, ) from temporalio.contrib.workflow_streams import WorkflowStreamClient from temporalio.exceptions import ApplicationError @@ -225,11 +224,10 @@ async def _empty_on_invoke_handoff(_ctx: RunContextWrapper[Any], _input: str) -> async def _noop_shell_executor(*_a: Any, **_kw: Any) -> str: - """Satisfies the ShellExecutor type for tool reconstruction during model calls.""" return "" -def _build_tool(tool: ToolInput) -> Tool: +def _build_tool(tool: ToolInput, env_refs: _WorkerEnvRefResolver) -> Tool: """Reconstruct a Tool from its data-conversion-friendly input form.""" if isinstance( tool, @@ -244,25 +242,22 @@ def _build_tool(tool: ToolInput) -> Tool: return tool elif isinstance(tool, CodeInterpreterTool): return CodeInterpreterTool( - tool_config=resolve_code_interpreter_tool_config(tool.tool_config) + tool_config=env_refs.resolve_code_interpreter_tool_config(tool.tool_config) ) elif isinstance(tool, ShellToolInput): - environment = resolve_shell_tool_environment(tool.environment) - # Only a local environment takes an executor, and an absent type means - # local, matching how ShellTool normalizes its environment. + environment = env_refs.resolve_shell_tool_environment(tool.environment) + # Only a local environment takes an executor. return ShellTool( name=tool.name, environment=environment, - executor=( - _noop_shell_executor - if environment.get("type", "local") == "local" - else None - ), + executor=_noop_shell_executor if environment["type"] == "local" else None, ) elif isinstance(tool, ApplyPatchToolInput): return ApplyPatchTool(name=tool.name, editor=_NoopApplyPatchEditor()) elif isinstance(tool, HostedMCPToolInput): - return HostedMCPTool(tool_config=resolve_mcp_tool_config(tool.tool_config)) + return HostedMCPTool( + tool_config=env_refs.resolve_mcp_tool_config(tool.tool_config) + ) elif isinstance(tool, CustomToolInput): return CustomTool( name=tool.tool_config["name"], @@ -285,8 +280,9 @@ def _build_tool(tool: ToolInput) -> Tool: def _build_tools_and_handoffs( input: ActivityModelInput, + env_refs: _WorkerEnvRefResolver, ) -> tuple[list[Tool], list[Handoff[Any, Any]]]: - tools = [_build_tool(x) for x in input.get("tools", [])] + tools = [_build_tool(x, env_refs) for x in input.get("tools", [])] handoffs: list[Handoff[Any, Any]] = [ Handoff( tool_name=x.tool_name, @@ -343,18 +339,23 @@ class ModelActivity: Disabling retries in your model of choice is recommended to allow activity retries to define the retry model. """ - def __init__(self, model_provider: ModelProvider | None = None): + def __init__( + self, + model_provider: ModelProvider | None = None, + resolvable_worker_env_vars: Collection[str] = (), + ): """Initialize the activity with a model provider.""" self._model_provider = model_provider or OpenAIProvider( openai_client=AsyncOpenAI(max_retries=0) ) + self._env_refs = _WorkerEnvRefResolver(resolvable_worker_env_vars) @activity.defn @auto_heartbeater async def invoke_model_activity(self, input: ActivityModelInput) -> ModelResponse: """Activity that invokes a model with the given input.""" model = self._model_provider.get_model(input.get("model_name")) - tools, handoffs = _build_tools_and_handoffs(input) + tools, handoffs = _build_tools_and_handoffs(input, self._env_refs) try: return await model.get_response( @@ -398,7 +399,7 @@ async def invoke_model_activity_streaming( trip ``heartbeat_timeout``. """ model = self._model_provider.get_model(input.get("model_name")) - tools, handoffs = _build_tools_and_handoffs(input) + tools, handoffs = _build_tools_and_handoffs(input, self._env_refs) topic = input["streaming_topic"] batch_interval = input.get( diff --git a/temporalio/contrib/openai_agents/_secret_reference.py b/temporalio/contrib/openai_agents/_secret_reference.py deleted file mode 100644 index cbf266047..000000000 --- a/temporalio/contrib/openai_agents/_secret_reference.py +++ /dev/null @@ -1,228 +0,0 @@ -"""References to secrets held in the worker process environment. - -A resolved secret is never written back into the activity's own input: the -resolvers copy at every level they write to, so the input keeps the marker. -""" - -from __future__ import annotations - -import os -from collections.abc import Iterator, Mapping, MutableMapping -from typing import Any, cast - -from agents.tool import ShellToolContainerAutoEnvironment, ShellToolEnvironment -from openai.types.responses.tool_param import CodeInterpreter, Mcp -from pydantic import ValidationError - -from temporalio.contrib.openai_agents.workflow import AgentsWorkflowError -from temporalio.exceptions import ApplicationError - -_MARKER_PREFIX = "temporal.secret_reference:" - -_ERROR_TYPE = "SecretReferenceFailure" - - -def secret_reference(key: str) -> str: - """Refer to a secret held in the worker process environment. - - .. warning:: - This function is experimental and may change in future versions. - Use with caution in production environments. - - Use it for a hosted tool credential that should come from the worker's - environment rather than being written into your workflow. Put the - placeholder returned where the credential would have gone. Only the - variable's name is recorded in workflow history. - - :: - - from agents import HostedMCPTool - - from temporalio.contrib.openai_agents import secret_reference - - tool = HostedMCPTool( - tool_config={ - "type": "mcp", - "server_label": "my_server", - "server_url": "https://example.com/mcp", - "authorization": secret_reference("MY_MCP_TOKEN"), - } - ) - - Set the variable on every worker that runs model activities — if it is - missing or empty there, the model call fails with a non-retryable - ``ApplicationError`` of type ``SecretReferenceFailure`` naming it. - - Args: - key: Name of the environment variable to read on the worker. - - Returns: - A placeholder string to use in place of the secret. - - Raises: - AgentsWorkflowError: If ``key`` is empty. - """ - if not key: - raise AgentsWorkflowError( - "secret_reference() requires the name of an environment variable to read " - "on the worker, but the name given was empty." - ) - return _MARKER_PREFIX + key - - -def _resolve_secret_reference(value: str) -> str: - """Return ``value`` with a secret reference marker replaced by its secret. - - Raises: - ApplicationError: If the marker names no environment variable, or the - variable it names is unset or empty in the worker process - environment. Non-retryable, of type ``SecretReferenceFailure``. - """ - if not value.startswith(_MARKER_PREFIX): - return value - key = value[len(_MARKER_PREFIX) :] - if not key: - raise ApplicationError( - f"Malformed secret reference {value!r}: the text after " - f"{_MARKER_PREFIX!r} must be the name of an environment variable to read " - "on the worker. Build the placeholder with secret_reference().", - type=_ERROR_TYPE, - non_retryable=True, - ) - secret = os.environ.get(key) - if not secret: - raise ApplicationError( - f"Secret reference environment variable {key!r} is not set, or is empty, " - "in the worker process environment.", - type=_ERROR_TYPE, - non_retryable=True, - ) - return secret - - -def _shallow_copy(mapping: Any) -> Any: - return dict(cast(Mapping[str, Any], mapping)) - - -def _malformed_domain_secret_error(e: ValidationError) -> ApplicationError: - """The rejection to raise for a domain secret that does not validate. - - pydantic rejects the whole entry for some malformed shapes and a single - field for others, so the type named is not claimed to be the entry's. - """ - error = e.errors()[0] - return ApplicationError( - f"Domain secret {error['loc'][0]} in a container network policy is " - f"malformed. Only its position and the type of the value that was " - f"rejected ({type(error['input']).__name__}) are reported: a malformed " - "entry could itself hold the secret.", - type=_ERROR_TYPE, - non_retryable=True, - ) - - -class _UnreadDomainSecrets: - """Raises when iterated, so secrets a failed read consumed never read as absent.""" - - def __init__(self, error: ApplicationError) -> None: - self._error = error - - def __iter__(self) -> Iterator[Any]: - raise self._error - - -def _resolve_network_policy(network_policy: Any) -> Any: - """Copy a container network policy, resolving each domain secret value. - - On the code interpreter path pydantic deserializes ``domain_secrets`` into a - single-pass iterator, so the entries read here go back onto the input. - - Raises: - ApplicationError: If a marker cannot be resolved, or a domain secret is - malformed. Non-retryable. - """ - policy = cast(MutableMapping[str, Any], network_policy) - domain_secrets = policy.get("domain_secrets") - if domain_secrets is None: - return dict(policy) - try: - unresolved = list(domain_secrets) - except ValidationError as e: - error = _malformed_domain_secret_error(e) - policy["domain_secrets"] = _UnreadDomainSecrets(error) - # Not chained: the validation error carries the entry it rejected. - raise error from None - policy["domain_secrets"] = unresolved - resolved = dict(policy) - resolved["domain_secrets"] = [ - _resolve_domain_secret(secret) for secret in unresolved - ] - return resolved - - -def _resolve_domain_secret(secret: Mapping[str, Any]) -> dict[str, Any]: - return {**secret, "value": _resolve_secret_reference(secret["value"])} - - -def resolve_mcp_tool_config(tool_config: Mcp) -> Mcp: - """Copy a hosted MCP tool config, resolving its authorization and headers. - - Raises: - ApplicationError: If a marker cannot be resolved. Non-retryable. - """ - resolved = _shallow_copy(tool_config) - if "authorization" in tool_config: - resolved["authorization"] = _resolve_secret_reference( - tool_config["authorization"] - ) - headers = tool_config.get("headers") - if headers is not None: - resolved["headers"] = { - name: _resolve_secret_reference(value) for name, value in headers.items() - } - return resolved - - -def resolve_shell_tool_environment( - environment: ShellToolEnvironment | None, -) -> ShellToolEnvironment: - """Copy a shell tool environment, resolving its domain secret values. - - An absent environment comes back as the local one ``ShellTool`` normalizes it to. - - Raises: - ApplicationError: If a marker cannot be resolved, or a domain secret is - malformed. Non-retryable. - """ - if environment is None: - return {"type": "local"} - if environment.get("type") != "container_auto": - return _shallow_copy(environment) - auto = cast(ShellToolContainerAutoEnvironment, environment) - network_policy = auto.get("network_policy") - resolved = _shallow_copy(auto) - if network_policy is not None: - resolved["network_policy"] = _resolve_network_policy(network_policy) - return resolved - - -def resolve_code_interpreter_tool_config( - tool_config: CodeInterpreter, -) -> CodeInterpreter: - """Copy a code interpreter tool config, resolving its domain secret values. - - Raises: - ApplicationError: If a marker cannot be resolved, or a domain secret is - malformed. Non-retryable. - """ - resolved = _shallow_copy(tool_config) - container = tool_config.get("container") - if not isinstance(container, Mapping): - return resolved - network_policy = container.get("network_policy") - if network_policy is None: - return resolved - resolved_container = _shallow_copy(container) - resolved_container["network_policy"] = _resolve_network_policy(network_policy) - resolved["container"] = resolved_container - return resolved diff --git a/temporalio/contrib/openai_agents/_temporal_openai_agents.py b/temporalio/contrib/openai_agents/_temporal_openai_agents.py index 43594657f..d4d91130e 100644 --- a/temporalio/contrib/openai_agents/_temporal_openai_agents.py +++ b/temporalio/contrib/openai_agents/_temporal_openai_agents.py @@ -3,7 +3,7 @@ import dataclasses import json import typing -from collections.abc import AsyncIterator, Callable, Iterator, Sequence +from collections.abc import AsyncIterator, Callable, Collection, Iterator, Sequence from contextlib import asynccontextmanager, contextmanager from datetime import timedelta @@ -219,6 +219,7 @@ def __init__( register_activities: bool = True, add_temporal_spans: bool = True, use_otel_instrumentation: bool = False, + resolvable_worker_env_vars: Collection[str] = (), ) -> None: """Initialize the OpenAI agents plugin. @@ -246,6 +247,12 @@ def __init__( use_otel_instrumentation: If set to true, enable open telemetry instrumentation. Warning: use_otel_instrumentation is experimental and behavior may change in future versions. Use with caution in production environments. + resolvable_worker_env_vars: Names of the environment variables that + ``temporal_worker_env_ref()`` may read on this worker. Names are + matched exactly, with no globbing or prefix matching; ``"*"`` + anywhere in the collection allows every name. + Warning: resolvable_worker_env_vars is experimental and behavior may change in future versions. + Use with caution in production environments. """ if model_params is None: @@ -275,7 +282,9 @@ def add_activities( if not register_activities: return activities or [] - model_activity = ModelActivity(model_provider) + model_activity = ModelActivity( + model_provider, resolvable_worker_env_vars=resolvable_worker_env_vars + ) new_activities = [ model_activity.invoke_model_activity, model_activity.invoke_model_activity_streaming, diff --git a/temporalio/contrib/openai_agents/_temporal_worker_env_ref.py b/temporalio/contrib/openai_agents/_temporal_worker_env_ref.py new file mode 100644 index 000000000..403c77314 --- /dev/null +++ b/temporalio/contrib/openai_agents/_temporal_worker_env_ref.py @@ -0,0 +1,153 @@ +"""References to secrets held in the Temporal Worker's environment.""" + +from __future__ import annotations + +import os +import re +from collections.abc import Collection, Mapping, MutableMapping +from typing import Any, cast + +from agents.tool import ShellToolContainerAutoEnvironment, ShellToolEnvironment +from openai.types.responses.tool_param import CodeInterpreter, Mcp + +_REF_PREFIX = "temporal.worker_env_ref:" + +_REF_PATTERN = re.compile(re.escape(_REF_PREFIX) + r"\{([^}{]*)\}") + +_ANY_ENV_VAR = "*" + + +def temporal_worker_env_ref(name: str) -> str: + """Refer to a secret held in the Temporal Worker's environment. + + .. warning:: + This function is experimental and may change in future versions. + Use with caution in production environments. + + Use it for a hosted tool credential that should come from the worker's + environment rather than being written into your workflow. Put the reference + returned in a ``HostedMCPTool``'s ``authorization`` or header value, or in the + ``value`` of a ``domain_secrets`` entry under a ``ShellTool`` or + ``CodeInterpreterTool``; the reference carries only the variable's name, never + its value. + + :: + + from agents import HostedMCPTool + + from temporalio.contrib.openai_agents import temporal_worker_env_ref + + tool = HostedMCPTool( + tool_config={ + "type": "mcp", + "server_label": "my_server", + "server_url": "https://example.com/mcp", + "authorization": temporal_worker_env_ref("MY_MCP_TOKEN"), + } + ) + + Every worker that runs model activities must set the variable and name it in + ``OpenAIAgentsPlugin(resolvable_worker_env_vars=["MY_MCP_TOKEN"])``. A name the + worker does not allow is sent to the model provider as literal text; an allowed + name that is unset resolves to an empty value. + + Args: + name: Name of the environment variable to read on the worker. + + Returns: + A reference string to use in place of the secret. + """ + return f"{_REF_PREFIX}{{{name}}}" + + +class _WorkerEnvRefResolver: # type:ignore[reportUnusedClass] + def __init__(self, resolvable_worker_env_vars: Collection[str]) -> None: + if isinstance(resolvable_worker_env_vars, str): + raise TypeError( + "resolvable_worker_env_vars takes a collection of environment variable " + 'names, such as ["MY_MCP_TOKEN"]. A single string is read as the ' + "collection of its characters, so pass a list even for one name." + ) + self._allowed = frozenset(resolvable_worker_env_vars) + + def _resolve_ref(self, value: str) -> str: + def substitute(match: re.Match[str]) -> str: + name = match.group(1) + if _ANY_ENV_VAR not in self._allowed and name not in self._allowed: + return match.group(0) + return os.environ.get(name, "") + + return _REF_PATTERN.sub(substitute, value) + + def _resolve_domain_secret(self, secret: Mapping[str, Any]) -> dict[str, Any]: + return {**secret, "value": self._resolve_ref(secret["value"])} + + def _resolve_network_policy(self, network_policy: Any) -> Any: + """On the code interpreter path pydantic deserializes ``domain_secrets`` into a + single-pass iterator, so the entries read here go back onto the input. + """ + policy = cast(MutableMapping[str, Any], network_policy) + domain_secrets = policy.get("domain_secrets") + if domain_secrets is None: + return network_policy + unresolved = list(domain_secrets) + policy["domain_secrets"] = unresolved + return { + **policy, + "domain_secrets": [ + self._resolve_domain_secret(secret) for secret in unresolved + ], + } + + def resolve_mcp_tool_config(self, tool_config: Mcp) -> Mcp: + resolved: Mcp = tool_config + if "authorization" in resolved: + resolved = { + **resolved, + "authorization": self._resolve_ref(resolved["authorization"]), + } + headers = resolved.get("headers") + if headers is not None: + resolved = { + **resolved, + "headers": { + name: self._resolve_ref(value) for name, value in headers.items() + }, + } + return resolved + + def resolve_shell_tool_environment( + self, + environment: ShellToolEnvironment | None, + ) -> ShellToolEnvironment: + """An absent environment comes back as the local one ``ShellTool`` normalizes it to.""" + if environment is None: + return {"type": "local"} + if environment.get("type") != "container_auto": + return environment + auto = cast(ShellToolContainerAutoEnvironment, environment) + network_policy = auto.get("network_policy") + if network_policy is None: + return environment + return { + **auto, + "network_policy": self._resolve_network_policy(network_policy), + } + + def resolve_code_interpreter_tool_config( + self, + tool_config: CodeInterpreter, + ) -> CodeInterpreter: + container = tool_config.get("container") + if not isinstance(container, Mapping): + return tool_config + network_policy = container.get("network_policy") + if network_policy is None: + return tool_config + return { + **tool_config, + "container": { + **container, + "network_policy": self._resolve_network_policy(network_policy), + }, + } diff --git a/temporalio/contrib/openai_agents/testing.py b/temporalio/contrib/openai_agents/testing.py index 110ca20b7..05d7864d5 100644 --- a/temporalio/contrib/openai_agents/testing.py +++ b/temporalio/contrib/openai_agents/testing.py @@ -1,6 +1,6 @@ """Testing utilities for OpenAI agents.""" -from collections.abc import AsyncIterator, Callable, Sequence +from collections.abc import AsyncIterator, Callable, Collection, Sequence from typing import Any from agents import ( @@ -181,6 +181,7 @@ def __init__( register_activities: bool = True, add_temporal_spans: bool = True, use_otel_instrumentation: bool = False, + resolvable_worker_env_vars: Collection[str] = (), ) -> None: """Initialize the AgentEnvironment. @@ -201,6 +202,10 @@ def __init__( use_otel_instrumentation: If set to true, enable open telemetry instrumentation. Warning: use_otel_instrumentation is experimental and behavior may change in future versions. Use with caution in production environments. + resolvable_worker_env_vars: Names of the environment variables that + ``temporal_worker_env_ref()`` may read on this environment's workers. + Warning: resolvable_worker_env_vars is experimental and behavior may change in future versions. + Use with caution in production environments. """ self._model_params = model_params self._model_provider = None @@ -213,6 +218,7 @@ def __init__( self._plugin: OpenAIAgentsPlugin | None = None self._add_temporal_spans = add_temporal_spans self._use_otel_instrumentation = use_otel_instrumentation + self._resolvable_worker_env_vars = resolvable_worker_env_vars async def __aenter__(self) -> "AgentEnvironment": """Enter the async context manager.""" @@ -224,6 +230,7 @@ async def __aenter__(self) -> "AgentEnvironment": register_activities=self._register_activities, add_temporal_spans=self._add_temporal_spans, use_otel_instrumentation=self._use_otel_instrumentation, + resolvable_worker_env_vars=self._resolvable_worker_env_vars, ) return self diff --git a/tests/contrib/openai_agents/test_openai.py b/tests/contrib/openai_agents/test_openai.py index 25597ee55..79f7f9dcd 100644 --- a/tests/contrib/openai_agents/test_openai.py +++ b/tests/contrib/openai_agents/test_openai.py @@ -96,6 +96,9 @@ from temporalio.contrib.openai_agents._temporal_model_stub import ( _TemporalModelStub, ) +from temporalio.contrib.openai_agents._temporal_worker_env_ref import ( + _WorkerEnvRefResolver, +) from temporalio.contrib.openai_agents.testing import ( AgentEnvironment, ResponseBuilders, @@ -2713,7 +2716,7 @@ class FakeSandboxSession: tool_inputs = activity_input.get("tools") or [] assert len(tool_inputs) == 1 - rebuilt = _build_tool(tool_inputs[0]) + rebuilt = _build_tool(tool_inputs[0], _WorkerEnvRefResolver(())) assert isinstance(rebuilt, CustomTool) assert rebuilt.name == tool.name assert rebuilt.description == tool.description @@ -2753,7 +2756,7 @@ async def stub(_ctx: Any, _payload: str) -> str: tool_inputs = activity_input.get("tools") or [] assert len(tool_inputs) == 1 - rebuilt = _build_tool(tool_inputs[0]) + rebuilt = _build_tool(tool_inputs[0], _WorkerEnvRefResolver(())) assert isinstance(rebuilt, CustomTool) assert rebuilt.tool_config == tool.tool_config assert rebuilt.defer_loading is True diff --git a/tests/contrib/openai_agents/test_openai_tool_secrets.py b/tests/contrib/openai_agents/test_openai_tool_secrets.py index 7689c0d0e..de4a3e0c4 100644 --- a/tests/contrib/openai_agents/test_openai_tool_secrets.py +++ b/tests/contrib/openai_agents/test_openai_tool_secrets.py @@ -1,11 +1,13 @@ -"""Tests for secret references in hosted tool secrets.""" +"""Tests for worker environment references in hosted tool secrets.""" +import time import uuid -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Collection from typing import Any, cast import pytest from agents import ( + Agent, AgentOutputSchemaBase, CodeInterpreterTool, Handoff, @@ -14,6 +16,7 @@ ModelResponse, ModelSettings, ModelTracing, + Runner, Tool, TResponseInputItem, Usage, @@ -22,13 +25,12 @@ from agents.tool import ShellTool, ShellToolEnvironment from temporalio import workflow -from temporalio.api.failure.v1 import Failure from temporalio.client import Client from temporalio.contrib.openai_agents import ( - AgentsWorkflowError, ModelActivityParameters, + OpenAIAgentsPlugin, OpenAIPayloadConverter, - secret_reference, + temporal_worker_env_ref, ) from temporalio.contrib.openai_agents._invoke_model_activity import ( ActivityModelInput, @@ -37,29 +39,25 @@ _build_tool, ) from temporalio.contrib.openai_agents._temporal_model_stub import _TemporalModelStub +from temporalio.contrib.openai_agents._temporal_worker_env_ref import ( + _WorkerEnvRefResolver, +) from temporalio.contrib.openai_agents.testing import ( AgentEnvironment, - TestModel, TestModelProvider, ) -from temporalio.converter import DefaultFailureConverter, PayloadConverter -from temporalio.exceptions import ApplicationError from temporalio.testing import ActivityEnvironment from tests.helpers import new_worker -# Fabricated secrets. Neither may reach a serialized activity argument. SENTINEL = "sk-test-sentinel-4f1a9c7e2b" -ENV_KEY = "TEMPORAL_TEST_TOOL_SECRET" +ENV_NAME = "TEMPORAL_TEST_TOOL_SECRET" OTHER_SENTINEL = "sk-test-other-8c3d5e0a1f" -OTHER_ENV_KEY = "TEMPORAL_TEST_OTHER_TOOL_SECRET" +OTHER_ENV_NAME = "TEMPORAL_TEST_OTHER_TOOL_SECRET" +_RESOLVER_ALLOWING_TEST_NAMES = _WorkerEnvRefResolver([ENV_NAME, OTHER_ENV_NAME]) -def _round_trip_activity_input(tool: Tool) -> tuple[bytes, ActivityModelInput]: - """Serialize the activity arguments a workflow would send for ``tool``. - Returns the serialized payload bytes and the input as the activity receives - it after deserialization. - """ +def _round_trip_activity_input(tool: Tool) -> tuple[bytes, ActivityModelInput]: stub = _TemporalModelStub( model_name="gpt-5", model_params=ModelActivityParameters(), @@ -83,7 +81,6 @@ def _round_trip_activity_input(tool: Tool) -> tuple[bytes, ActivityModelInput]: def _activity_input_payload_and_tool(tool: Tool) -> tuple[bytes, Any]: - """The serialized payload, and the single tool the activity receives.""" payload, received = _round_trip_activity_input(tool) tools = received.get("tools") or [] assert len(tools) == 1 @@ -106,7 +103,7 @@ def _domain_secret(name: str, value: str) -> dict[str, str]: return {"domain": "example.com", "name": name, "value": value} -def _allowlist(domain_secrets: tuple[Any, ...]) -> Any: +def _network_policy(domain_secrets: tuple[Any, ...]) -> Any: return { "type": "allowlist", "allowed_domains": ["example.com"], @@ -117,7 +114,7 @@ def _allowlist(domain_secrets: tuple[Any, ...]) -> Any: def _shell_tool(*domain_secrets: Any) -> ShellTool: environment: Any = { "type": "container_auto", - "network_policy": _allowlist(domain_secrets), + "network_policy": _network_policy(domain_secrets), } return ShellTool(environment=environment) @@ -127,125 +124,155 @@ def _code_interpreter_tool(*domain_secrets: Any) -> CodeInterpreterTool: "type": "code_interpreter", "container": { "type": "auto", - "network_policy": _allowlist(domain_secrets), + "network_policy": _network_policy(domain_secrets), }, } return CodeInterpreterTool(tool_config=tool_config) -def _fields(value: Any) -> dict[str, Any]: - """View a TypedDict-shaped value as a plain mapping, for assertions.""" +def _as_dict(value: Any) -> dict[str, Any]: return cast(dict[str, Any], value) -def _domain_secrets(network_policy: Any) -> list[Any]: - return list(_fields(network_policy)["domain_secrets"]) - - -def _reported_failure(error: BaseException) -> Failure: - """The failure a worker would report to the server for ``error``. - - The converter walks ``__cause__``, or the implicit ``__context__`` when - there is none, into ``failure.cause``. - """ - failure = Failure() - DefaultFailureConverter().to_failure(error, PayloadConverter.default, failure) - return failure +def _secrets_in(network_policy: Any) -> list[Any]: + return list(_as_dict(network_policy)["domain_secrets"]) def _shell_secrets(built: Tool) -> list[Any]: assert isinstance(built, ShellTool) assert built.environment is not None - return _domain_secrets(_fields(built.environment)["network_policy"]) + return _secrets_in(_as_dict(built.environment)["network_policy"]) def _code_interpreter_secrets(built: Tool) -> list[Any]: assert isinstance(built, CodeInterpreterTool) - container = _fields(built.tool_config)["container"] - return _domain_secrets(_fields(container)["network_policy"]) + container = _as_dict(built.tool_config)["container"] + return _secrets_in(_as_dict(container)["network_policy"]) def test_hosted_mcp_secrets_stay_out_of_activity_arguments( monkeypatch: pytest.MonkeyPatch, ): - monkeypatch.setenv(ENV_KEY, SENTINEL) - marker = secret_reference(ENV_KEY) + monkeypatch.setenv(ENV_NAME, SENTINEL) + ref = temporal_worker_env_ref(ENV_NAME) - payload, received = _activity_input_payload_and_tool( - _hosted_mcp_tool(marker, marker) - ) + payload, received = _activity_input_payload_and_tool(_hosted_mcp_tool(ref, ref)) assert SENTINEL.encode() not in payload - assert payload.count(marker.encode()) == 2 - assert received.tool_config["authorization"] == marker - assert received.tool_config["headers"]["X-Token"] == marker + assert payload.count(ref.encode()) == 2 + assert received.tool_config["authorization"] == ref + assert received.tool_config["headers"]["X-Token"] == ref def test_hosted_shell_domain_secret_stays_out_of_activity_arguments( monkeypatch: pytest.MonkeyPatch, ): - monkeypatch.setenv(ENV_KEY, SENTINEL) - marker = secret_reference(ENV_KEY) + monkeypatch.setenv(ENV_NAME, SENTINEL) + ref = temporal_worker_env_ref(ENV_NAME) payload, received = _activity_input_payload_and_tool( - _shell_tool(_domain_secret("TOKEN", marker)) + _shell_tool(_domain_secret("TOKEN", ref)) ) assert SENTINEL.encode() not in payload - assert marker.encode() in payload - secrets = _domain_secrets(received.environment["network_policy"]) - assert secrets[0]["value"] == marker + assert ref.encode() in payload + secrets = _secrets_in(received.environment["network_policy"]) + assert secrets[0]["value"] == ref def test_code_interpreter_domain_secret_stays_out_of_activity_arguments( monkeypatch: pytest.MonkeyPatch, ): - monkeypatch.setenv(ENV_KEY, SENTINEL) - marker = secret_reference(ENV_KEY) + monkeypatch.setenv(ENV_NAME, SENTINEL) + ref = temporal_worker_env_ref(ENV_NAME) payload, received = _activity_input_payload_and_tool( - _code_interpreter_tool(_domain_secret("TOKEN", marker)) + _code_interpreter_tool(_domain_secret("TOKEN", ref)) ) assert SENTINEL.encode() not in payload - assert marker.encode() in payload - secrets = _domain_secrets(received.tool_config["container"]["network_policy"]) - assert secrets[0]["value"] == marker + assert ref.encode() in payload + secrets = _secrets_in(received.tool_config["container"]["network_policy"]) + assert secrets[0]["value"] == ref def test_hosted_mcp_secrets_resolve_for_the_model_call( monkeypatch: pytest.MonkeyPatch, ): - monkeypatch.setenv(ENV_KEY, SENTINEL) - marker = secret_reference(ENV_KEY) - _payload, received = _activity_input_payload_and_tool( - _hosted_mcp_tool(marker, marker) - ) + monkeypatch.setenv(ENV_NAME, SENTINEL) + ref = temporal_worker_env_ref(ENV_NAME) + _payload, received = _activity_input_payload_and_tool(_hosted_mcp_tool(ref, ref)) - built = _build_tool(received) + built = _build_tool(received, _RESOLVER_ALLOWING_TEST_NAMES) assert isinstance(built, HostedMCPTool) - assert _fields(built.tool_config)["authorization"] == SENTINEL - assert _fields(built.tool_config)["headers"] == { + assert _as_dict(built.tool_config)["authorization"] == SENTINEL + assert _as_dict(built.tool_config)["headers"] == { "X-Token": SENTINEL, "X-Plain": "not-a-secret", } - # The deserialized activity argument still holds the marker, so building - # again resolves the same secret rather than an emptied config. - assert received.tool_config["authorization"] == marker - assert received.tool_config["headers"]["X-Token"] == marker + assert received.tool_config["authorization"] == ref + assert received.tool_config["headers"]["X-Token"] == ref + + +@pytest.mark.parametrize( + ("resolvable", "resolves"), + [([ENV_NAME], True), (["*"], True), ([OTHER_ENV_NAME], False)], + ids=["the_name", "star", "another_name"], +) +def test_a_reference_resolves_only_from_a_variable_the_worker_allows( + monkeypatch: pytest.MonkeyPatch, resolvable: Collection[str], resolves: bool +): + monkeypatch.setenv(ENV_NAME, SENTINEL) + ref = temporal_worker_env_ref(ENV_NAME) + _payload, received = _activity_input_payload_and_tool(_hosted_mcp_tool(ref, ref)) + + built = _build_tool(received, _WorkerEnvRefResolver(resolvable)) + + assert isinstance(built, HostedMCPTool) + expected = SENTINEL if resolves else ref + assert _as_dict(built.tool_config)["authorization"] == expected + assert _as_dict(built.tool_config)["headers"]["X-Token"] == expected + + +def test_star_anywhere_in_the_resolvable_names_resolves_every_name( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv(ENV_NAME, SENTINEL) + ref = temporal_worker_env_ref(ENV_NAME) + _payload, received = _activity_input_payload_and_tool(_hosted_mcp_tool(ref, ref)) + + built = _build_tool(received, _WorkerEnvRefResolver([OTHER_ENV_NAME, "*"])) + + assert isinstance(built, HostedMCPTool) + assert _as_dict(built.tool_config)["authorization"] == SENTINEL + assert _as_dict(built.tool_config)["headers"]["X-Token"] == SENTINEL + + +def test_a_glob_in_the_resolvable_names_matches_no_name( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv(ENV_NAME, SENTINEL) + ref = temporal_worker_env_ref(ENV_NAME) + _payload, received = _activity_input_payload_and_tool(_hosted_mcp_tool(ref, ref)) + + built = _build_tool(received, _WorkerEnvRefResolver(["TEMPORAL_TEST_*"])) + + assert isinstance(built, HostedMCPTool) + assert _as_dict(built.tool_config)["authorization"] == ref + assert _as_dict(built.tool_config)["headers"]["X-Token"] == ref def test_hosted_shell_domain_secret_resolves_for_the_model_call( monkeypatch: pytest.MonkeyPatch, ): - monkeypatch.setenv(ENV_KEY, SENTINEL) - marker = secret_reference(ENV_KEY) + monkeypatch.setenv(ENV_NAME, SENTINEL) + ref = temporal_worker_env_ref(ENV_NAME) _payload, received = _activity_input_payload_and_tool( - _shell_tool(_domain_secret("TOKEN", marker)) + _shell_tool(_domain_secret("TOKEN", ref)) ) - built = _build_tool(received) + built = _build_tool(received, _RESOLVER_ALLOWING_TEST_NAMES) assert _shell_secrets(built) == [_domain_secret("TOKEN", SENTINEL)] @@ -253,13 +280,13 @@ def test_hosted_shell_domain_secret_resolves_for_the_model_call( def test_code_interpreter_domain_secret_resolves_for_the_model_call( monkeypatch: pytest.MonkeyPatch, ): - monkeypatch.setenv(ENV_KEY, SENTINEL) - marker = secret_reference(ENV_KEY) + monkeypatch.setenv(ENV_NAME, SENTINEL) + ref = temporal_worker_env_ref(ENV_NAME) _payload, received = _activity_input_payload_and_tool( - _code_interpreter_tool(_domain_secret("TOKEN", marker)) + _code_interpreter_tool(_domain_secret("TOKEN", ref)) ) - built = _build_tool(received) + built = _build_tool(received, _RESOLVER_ALLOWING_TEST_NAMES) assert _code_interpreter_secrets(built) == [_domain_secret("TOKEN", SENTINEL)] @@ -267,56 +294,33 @@ def test_code_interpreter_domain_secret_resolves_for_the_model_call( def test_code_interpreter_domain_secret_survives_a_second_build( monkeypatch: pytest.MonkeyPatch, ): - """``domain_secrets`` deserializes into a single-pass iterator here. - - Building twice from one deserialized input must resolve the secret both - times, and must leave the input itself holding the marker. - """ - monkeypatch.setenv(ENV_KEY, SENTINEL) - marker = secret_reference(ENV_KEY) + monkeypatch.setenv(ENV_NAME, SENTINEL) + ref = temporal_worker_env_ref(ENV_NAME) _payload, received = _activity_input_payload_and_tool( - _code_interpreter_tool(_domain_secret("TOKEN", marker)) + _code_interpreter_tool(_domain_secret("TOKEN", ref)) ) - first = _build_tool(received) - second = _build_tool(received) + first = _build_tool(received, _RESOLVER_ALLOWING_TEST_NAMES) + second = _build_tool(received, _RESOLVER_ALLOWING_TEST_NAMES) assert _code_interpreter_secrets(first) == [_domain_secret("TOKEN", SENTINEL)] assert _code_interpreter_secrets(second) == [_domain_secret("TOKEN", SENTINEL)] - assert _domain_secrets(received.tool_config["container"]["network_policy"]) == [ - _domain_secret("TOKEN", marker) + assert _secrets_in(received.tool_config["container"]["network_policy"]) == [ + _domain_secret("TOKEN", ref) ] -def test_hosted_shell_domain_secret_survives_a_second_build( +def test_only_domain_secrets_holding_a_worker_env_ref_are_resolved( monkeypatch: pytest.MonkeyPatch, ): - monkeypatch.setenv(ENV_KEY, SENTINEL) - marker = secret_reference(ENV_KEY) - _payload, received = _activity_input_payload_and_tool( - _shell_tool(_domain_secret("TOKEN", marker)) - ) - - first = _build_tool(received) - second = _build_tool(received) - - assert _shell_secrets(first) == [_domain_secret("TOKEN", SENTINEL)] - assert _shell_secrets(second) == [_domain_secret("TOKEN", SENTINEL)] - assert _domain_secrets(received.environment["network_policy"]) == [ - _domain_secret("TOKEN", marker) - ] - - -def test_only_marker_domain_secrets_are_resolved(monkeypatch: pytest.MonkeyPatch): - """A literal value alongside a marker is left exactly as the workflow sent it.""" - monkeypatch.setenv(ENV_KEY, SENTINEL) - marker = secret_reference(ENV_KEY) + monkeypatch.setenv(ENV_NAME, SENTINEL) + ref = temporal_worker_env_ref(ENV_NAME) literal = _domain_secret("PLAIN", "plain-token-value") _payload, received = _activity_input_payload_and_tool( - _code_interpreter_tool(literal, _domain_secret("TOKEN", marker)) + _code_interpreter_tool(literal, _domain_secret("TOKEN", ref)) ) - built = _build_tool(received) + built = _build_tool(received, _RESOLVER_ALLOWING_TEST_NAMES) assert _code_interpreter_secrets(built) == [ literal, @@ -324,40 +328,19 @@ def test_only_marker_domain_secrets_are_resolved(monkeypatch: pytest.MonkeyPatch ] -def test_two_secret_references_in_one_mcp_config_resolve_to_their_own_secrets( - monkeypatch: pytest.MonkeyPatch, -): - """Each reference names its own variable, and stands for that one only.""" - monkeypatch.setenv(ENV_KEY, SENTINEL) - monkeypatch.setenv(OTHER_ENV_KEY, OTHER_SENTINEL) - _payload, received = _activity_input_payload_and_tool( - _hosted_mcp_tool(secret_reference(ENV_KEY), secret_reference(OTHER_ENV_KEY)) - ) - - built = _build_tool(received) - - assert isinstance(built, HostedMCPTool) - assert _fields(built.tool_config)["authorization"] == SENTINEL - assert _fields(built.tool_config)["headers"] == { - "X-Token": OTHER_SENTINEL, - "X-Plain": "not-a-secret", - } - - def test_two_domain_secrets_resolve_to_their_own_secrets( monkeypatch: pytest.MonkeyPatch, ): - """The entries are resolved one by one, each from the variable it names.""" - monkeypatch.setenv(ENV_KEY, SENTINEL) - monkeypatch.setenv(OTHER_ENV_KEY, OTHER_SENTINEL) + monkeypatch.setenv(ENV_NAME, SENTINEL) + monkeypatch.setenv(OTHER_ENV_NAME, OTHER_SENTINEL) _payload, received = _activity_input_payload_and_tool( _code_interpreter_tool( - _domain_secret("TOKEN", secret_reference(ENV_KEY)), - _domain_secret("OTHER", secret_reference(OTHER_ENV_KEY)), + _domain_secret("TOKEN", temporal_worker_env_ref(ENV_NAME)), + _domain_secret("OTHER", temporal_worker_env_ref(OTHER_ENV_NAME)), ) ) - built = _build_tool(received) + built = _build_tool(received, _RESOLVER_ALLOWING_TEST_NAMES) assert _code_interpreter_secrets(built) == [ _domain_secret("TOKEN", SENTINEL), @@ -365,20 +348,21 @@ def test_two_domain_secrets_resolve_to_their_own_secrets( ] -def test_shell_domain_secrets_resolve_by_position(monkeypatch: pytest.MonkeyPatch): - """Each entry resolves from the variable it names, whichever position it holds.""" - monkeypatch.setenv(ENV_KEY, SENTINEL) - monkeypatch.setenv(OTHER_ENV_KEY, OTHER_SENTINEL) +def test_shell_domain_secrets_resolve_to_their_own_secrets_in_order( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv(ENV_NAME, SENTINEL) + monkeypatch.setenv(OTHER_ENV_NAME, OTHER_SENTINEL) literal = _domain_secret("PLAIN", "plain-token-value") _payload, received = _activity_input_payload_and_tool( _shell_tool( literal, - _domain_secret("TOKEN", secret_reference(ENV_KEY)), - _domain_secret("OTHER", secret_reference(OTHER_ENV_KEY)), + _domain_secret("TOKEN", temporal_worker_env_ref(ENV_NAME)), + _domain_secret("OTHER", temporal_worker_env_ref(OTHER_ENV_NAME)), ) ) - built = _build_tool(received) + built = _build_tool(received, _RESOLVER_ALLOWING_TEST_NAMES) assert _shell_secrets(built) == [ literal, @@ -387,74 +371,120 @@ def test_shell_domain_secrets_resolve_by_position(monkeypatch: pytest.MonkeyPatc ] -def test_a_marker_in_a_header_name_is_passed_through( +def test_a_shell_domain_secret_naming_a_denied_variable_is_passed_through( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv(OTHER_ENV_NAME, OTHER_SENTINEL) + denied = temporal_worker_env_ref(OTHER_ENV_NAME) + _payload, received = _activity_input_payload_and_tool( + _shell_tool(_domain_secret("TOKEN", denied)) + ) + + built = _build_tool(received, _WorkerEnvRefResolver([ENV_NAME])) + + assert _shell_secrets(built) == [_domain_secret("TOKEN", denied)] + + +def test_a_code_interpreter_domain_secret_naming_a_denied_variable_is_passed_through( monkeypatch: pytest.MonkeyPatch, ): - """A marker belongs where a credential belongs, and a header name is not that.""" - monkeypatch.setenv(ENV_KEY, SENTINEL) - marker = secret_reference(ENV_KEY) + monkeypatch.setenv(OTHER_ENV_NAME, OTHER_SENTINEL) + denied = temporal_worker_env_ref(OTHER_ENV_NAME) + _payload, received = _activity_input_payload_and_tool( + _code_interpreter_tool(_domain_secret("TOKEN", denied)) + ) + + built = _build_tool(received, _WorkerEnvRefResolver([ENV_NAME])) + + assert _code_interpreter_secrets(built) == [_domain_secret("TOKEN", denied)] + + +def test_two_worker_env_refs_in_one_mcp_config_resolve_to_their_own_secrets( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv(ENV_NAME, SENTINEL) + monkeypatch.setenv(OTHER_ENV_NAME, OTHER_SENTINEL) + _payload, received = _activity_input_payload_and_tool( + _hosted_mcp_tool( + temporal_worker_env_ref(ENV_NAME), temporal_worker_env_ref(OTHER_ENV_NAME) + ) + ) + + built = _build_tool(received, _RESOLVER_ALLOWING_TEST_NAMES) + + assert isinstance(built, HostedMCPTool) + assert _as_dict(built.tool_config)["authorization"] == SENTINEL + assert _as_dict(built.tool_config)["headers"] == { + "X-Token": OTHER_SENTINEL, + "X-Plain": "not-a-secret", + } + + +def test_a_worker_env_ref_in_a_header_name_is_passed_through( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv(ENV_NAME, SENTINEL) + ref = temporal_worker_env_ref(ENV_NAME) tool_config: Any = { "type": "mcp", "server_label": "test_server", "server_url": "https://example.com/mcp", - "headers": {marker: "not-a-secret"}, + "headers": {ref: "not-a-secret"}, } payload, received = _activity_input_payload_and_tool( HostedMCPTool(tool_config=tool_config) ) - built = _build_tool(received) + built = _build_tool(received, _RESOLVER_ALLOWING_TEST_NAMES) assert isinstance(built, HostedMCPTool) - assert _fields(built.tool_config)["headers"] == {marker: "not-a-secret"} + assert _as_dict(built.tool_config)["headers"] == {ref: "not-a-secret"} assert SENTINEL.encode() not in payload def test_an_mcp_config_without_an_authorization_resolves_its_headers( monkeypatch: pytest.MonkeyPatch, ): - """``authorization`` is optional, and an absent one is not one to resolve.""" - monkeypatch.setenv(ENV_KEY, SENTINEL) + monkeypatch.setenv(ENV_NAME, SENTINEL) tool_config: Any = { "type": "mcp", "server_label": "test_server", "server_url": "https://example.com/mcp", - "headers": {"X-Token": secret_reference(ENV_KEY)}, + "headers": {"X-Token": temporal_worker_env_ref(ENV_NAME)}, } _payload, received = _activity_input_payload_and_tool( HostedMCPTool(tool_config=tool_config) ) - built = _build_tool(received) + built = _build_tool(received, _RESOLVER_ALLOWING_TEST_NAMES) assert isinstance(built, HostedMCPTool) - assert "authorization" not in _fields(built.tool_config) - assert _fields(built.tool_config)["headers"] == {"X-Token": SENTINEL} + assert "authorization" not in _as_dict(built.tool_config) + assert _as_dict(built.tool_config)["headers"] == {"X-Token": SENTINEL} def test_an_mcp_config_without_headers_resolves_its_authorization( monkeypatch: pytest.MonkeyPatch, ): - """``headers`` is optional, and an absent one is not one to resolve.""" - monkeypatch.setenv(ENV_KEY, SENTINEL) + monkeypatch.setenv(ENV_NAME, SENTINEL) tool_config: Any = { "type": "mcp", "server_label": "test_server", "server_url": "https://example.com/mcp", - "authorization": secret_reference(ENV_KEY), + "authorization": temporal_worker_env_ref(ENV_NAME), } _payload, received = _activity_input_payload_and_tool( HostedMCPTool(tool_config=tool_config) ) - built = _build_tool(received) + built = _build_tool(received, _RESOLVER_ALLOWING_TEST_NAMES) assert isinstance(built, HostedMCPTool) - assert _fields(built.tool_config)["authorization"] == SENTINEL - assert "headers" not in _fields(built.tool_config) + assert _as_dict(built.tool_config)["authorization"] == SENTINEL + assert "headers" not in _as_dict(built.tool_config) def test_local_shell_environment_keeps_its_executor(): @@ -462,11 +492,11 @@ def test_local_shell_environment_keeps_its_executor(): ShellTool(environment={"type": "local"}, executor=lambda _request: "") ) - built = _build_tool(received) + built = _build_tool(received, _RESOLVER_ALLOWING_TEST_NAMES) assert isinstance(built, ShellTool) assert built.executor is not None - assert _fields(built.environment) == {"type": "local"} + assert _as_dict(built.environment) == {"type": "local"} @pytest.mark.parametrize( @@ -484,30 +514,28 @@ def test_local_shell_environment_keeps_its_executor(): def test_hosted_shell_environment_gets_no_executor( environment: ShellToolEnvironment, ): - """A hosted environment runs on OpenAI's side, and rejects an executor.""" _payload, received = _activity_input_payload_and_tool( ShellTool(environment=environment) ) - built = _build_tool(received) + built = _build_tool(received, _RESOLVER_ALLOWING_TEST_NAMES) assert isinstance(built, ShellTool) assert built.executor is None - assert _fields(built.environment) == environment + assert _as_dict(built.environment) == environment def test_code_interpreter_container_id_is_passed_through(): - """A container named by ID carries no network policy, so it has no secrets.""" _payload, received = _activity_input_payload_and_tool( CodeInterpreterTool( tool_config={"type": "code_interpreter", "container": "cntr_abc"} ) ) - built = _build_tool(received) + built = _build_tool(received, _RESOLVER_ALLOWING_TEST_NAMES) assert isinstance(built, CodeInterpreterTool) - assert _fields(built.tool_config)["container"] == "cntr_abc" + assert _as_dict(built.tool_config)["container"] == "cntr_abc" @pytest.mark.parametrize( @@ -521,177 +549,151 @@ def test_code_interpreter_container_id_is_passed_through(): def test_code_interpreter_container_without_domain_secrets_is_passed_through( container: Any, ): - """Only a policy carries domain secrets, and only an allowlist has any.""" _payload, received = _activity_input_payload_and_tool( CodeInterpreterTool( tool_config={"type": "code_interpreter", "container": container} ) ) - built = _build_tool(received) + built = _build_tool(received, _RESOLVER_ALLOWING_TEST_NAMES) assert isinstance(built, CodeInterpreterTool) - assert _fields(built.tool_config)["container"] == container + assert _as_dict(built.tool_config)["container"] == container -@pytest.mark.parametrize("value", ["", None]) -def test_unset_or_empty_environment_variable_is_not_retryable( - monkeypatch: pytest.MonkeyPatch, value: str | None +def test_an_unset_environment_variable_resolves_to_an_empty_value( + monkeypatch: pytest.MonkeyPatch, ): - if value is None: - monkeypatch.delenv(ENV_KEY, raising=False) - else: - monkeypatch.setenv(ENV_KEY, value) - marker = secret_reference(ENV_KEY) - _payload, received = _activity_input_payload_and_tool( - _hosted_mcp_tool(marker, marker) - ) + monkeypatch.delenv(ENV_NAME, raising=False) + ref = temporal_worker_env_ref(ENV_NAME) + _payload, received = _activity_input_payload_and_tool(_hosted_mcp_tool(ref, ref)) - with pytest.raises(ApplicationError) as err: - _build_tool(received) + built = _build_tool(received, _RESOLVER_ALLOWING_TEST_NAMES) - assert err.value.non_retryable - assert err.value.type == "SecretReferenceFailure" - assert ENV_KEY in err.value.message + assert isinstance(built, HostedMCPTool) + assert _as_dict(built.tool_config)["authorization"] == "" + assert _as_dict(built.tool_config)["headers"]["X-Token"] == "" -def test_marker_without_a_variable_name_is_rejected_as_malformed(): - """The marker format is public, so a hand-written one can name nothing.""" +def test_a_policy_with_no_domain_secrets_is_passed_through(): + policy: Any = {"type": "allowlist", "allowed_domains": ["example.com"]} _payload, received = _activity_input_payload_and_tool( - _hosted_mcp_tool("temporal.secret_reference:", "plain") + CodeInterpreterTool( + tool_config={ + "type": "code_interpreter", + "container": {"type": "auto", "network_policy": policy}, + } + ) ) - with pytest.raises(ApplicationError) as err: - _build_tool(received) + built = _build_tool(received, _RESOLVER_ALLOWING_TEST_NAMES) - assert err.value.non_retryable - assert err.value.type == "SecretReferenceFailure" - assert "Malformed secret reference" in err.value.message + assert isinstance(built, CodeInterpreterTool) + assert _as_dict(built.tool_config)["container"] == { + "type": "auto", + "network_policy": policy, + } -@pytest.mark.parametrize( - ("entry", "type_name"), - [ - ({"domain": "example.com", "name": "TOKEN", "vaule": SENTINEL}, "dict"), - ({"domain": "example.com", "name": "TOKEN", "value": 7}, "int"), - (None, "NoneType"), - (42, "int"), - (SENTINEL, "str"), - ], - ids=["mis_keyed", "wrong_value_type", "null", "number", "bare_string"], -) -def test_a_malformed_domain_secret_is_rejected_non_retryably( - entry: Any, type_name: str +def test_plain_values_are_passed_through_unchanged( + monkeypatch: pytest.MonkeyPatch, ): - """The rejection reports a position and a type, and nothing else. - - Neither the message nor the failure a worker reports may carry what - pydantic rejected, which for some shapes is the whole entry. - """ - _payload, received = _activity_input_payload_and_tool(_code_interpreter_tool(entry)) + monkeypatch.setenv(ENV_NAME, SENTINEL) + plain = "temporal.worker_env_ref-but-not-quite" + _payload, received = _activity_input_payload_and_tool( + _hosted_mcp_tool(plain, plain) + ) - with pytest.raises(ApplicationError) as err: - _build_tool(received) + built = _build_tool(received, _RESOLVER_ALLOWING_TEST_NAMES) - assert err.value.non_retryable - assert err.value.type == "SecretReferenceFailure" - assert f"the type of the value that was rejected ({type_name})" in err.value.message - assert SENTINEL not in err.value.message - failure = _reported_failure(err.value) - assert not failure.HasField("cause") - assert SENTINEL.encode() not in failure.SerializeToString() + assert isinstance(built, HostedMCPTool) + assert _as_dict(built.tool_config)["authorization"] == plain + assert _as_dict(built.tool_config)["headers"] == { + "X-Token": plain, + "X-Plain": "not-a-secret", + } -def test_a_malformed_domain_secret_is_reported_by_position( +def test_a_worker_env_ref_inside_a_larger_value_is_substituted_in_place( monkeypatch: pytest.MonkeyPatch, ): - """A policy can carry several secrets, so the rejection says which one.""" - monkeypatch.setenv(ENV_KEY, SENTINEL) + monkeypatch.setenv(ENV_NAME, SENTINEL) + composed = "Bearer " + temporal_worker_env_ref(ENV_NAME) _payload, received = _activity_input_payload_and_tool( - _code_interpreter_tool(_domain_secret("TOKEN", secret_reference(ENV_KEY)), 42) + _hosted_mcp_tool(composed, composed) ) - with pytest.raises(ApplicationError) as err: - _build_tool(received) + built = _build_tool(received, _RESOLVER_ALLOWING_TEST_NAMES) - assert "Domain secret 1" in err.value.message - assert SENTINEL not in err.value.message + assert isinstance(built, HostedMCPTool) + assert _as_dict(built.tool_config)["authorization"] == f"Bearer {SENTINEL}" + assert _as_dict(built.tool_config)["headers"]["X-Token"] == f"Bearer {SENTINEL}" -def test_a_malformed_domain_secret_is_rejected_again_on_a_second_build(): - """A failed read consumes the secrets, so a second build has to fail too.""" - _payload, received = _activity_input_payload_and_tool(_code_interpreter_tool(42)) +def test_one_value_holding_two_refs_resolves_only_the_allowed_name( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv(ENV_NAME, SENTINEL) + monkeypatch.setenv(OTHER_ENV_NAME, OTHER_SENTINEL) + denied = temporal_worker_env_ref(OTHER_ENV_NAME) + composed = f"{temporal_worker_env_ref(ENV_NAME)} {denied}" + _payload, received = _activity_input_payload_and_tool( + _hosted_mcp_tool(composed, composed) + ) - with pytest.raises(ApplicationError): - _build_tool(received) - with pytest.raises(ApplicationError) as err: - _build_tool(received) + built = _build_tool(received, _WorkerEnvRefResolver([ENV_NAME])) - assert err.value.non_retryable - assert err.value.type == "SecretReferenceFailure" + assert isinstance(built, HostedMCPTool) + assert _as_dict(built.tool_config)["authorization"] == f"{SENTINEL} {denied}" + assert _as_dict(built.tool_config)["headers"]["X-Token"] == f"{SENTINEL} {denied}" -def test_a_policy_with_no_domain_secrets_is_passed_through(): - """No domain secrets is no secrets to resolve, and nothing to materialize.""" - policy: Any = {"type": "allowlist", "allowed_domains": ["example.com"]} +def test_a_worker_env_ref_with_no_closing_brace_is_passed_through( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv(ENV_NAME, SENTINEL) + truncated = "temporal.worker_env_ref:{" + ENV_NAME _payload, received = _activity_input_payload_and_tool( - CodeInterpreterTool( - tool_config={ - "type": "code_interpreter", - "container": {"type": "auto", "network_policy": policy}, - } - ) + _hosted_mcp_tool(truncated, truncated) ) - built = _build_tool(received) + built = _build_tool(received, _WorkerEnvRefResolver(["*"])) - assert isinstance(built, CodeInterpreterTool) - assert _fields(built.tool_config)["container"] == { - "type": "auto", - "network_policy": policy, - } + assert isinstance(built, HostedMCPTool) + assert _as_dict(built.tool_config)["authorization"] == truncated + assert _as_dict(built.tool_config)["headers"]["X-Token"] == truncated -def test_plain_values_are_passed_through_unchanged( - monkeypatch: pytest.MonkeyPatch, -): - monkeypatch.setenv(ENV_KEY, SENTINEL) - plain = "temporal.secret_reference-but-not-quite" +def test_a_value_packed_with_unclosed_references_does_not_stall_the_resolver(): + opener = "temporal.worker_env_ref:{" + packed = opener * (1024 * 1024 // len(opener)) _payload, received = _activity_input_payload_and_tool( - _hosted_mcp_tool(plain, plain) + _hosted_mcp_tool(packed, "not-a-secret") ) - built = _build_tool(received) + start = time.monotonic() + built = _build_tool(received, _WorkerEnvRefResolver([])) + elapsed = time.monotonic() - start assert isinstance(built, HostedMCPTool) - assert _fields(built.tool_config)["authorization"] == plain - assert _fields(built.tool_config)["headers"] == { - "X-Token": plain, - "X-Plain": "not-a-secret", - } + assert _as_dict(built.tool_config)["authorization"] == packed + assert elapsed < 5.0 -def test_secret_reference_rejects_an_empty_key(): - with pytest.raises(AgentsWorkflowError): - secret_reference("") +def test_a_bare_string_is_rejected_as_the_resolvable_variable_names(): + with pytest.raises(TypeError, match="resolvable_worker_env_vars"): + _WorkerEnvRefResolver(ENV_NAME) async def _no_stream_events() -> AsyncIterator[TResponseStreamEvent]: - """A stream that completes without events. - - Publishing one costs a real wait: the activity signals it to a workflow that - does not exist, and the flusher then retries for ten minutes. - """ + """Publishing an event here makes the flusher retry for ten minutes against a workflow that does not exist.""" events: list[TResponseStreamEvent] = [] for event in events: yield event class _ToolRecordingModel(Model): - """Records the tools the model activity hands the model.""" - def __init__(self) -> None: - self.called = False self.tools: list[Tool] = [] async def get_response( @@ -705,8 +707,6 @@ async def get_response( tracing: ModelTracing, **kwargs: Any, ) -> ModelResponse: - """Record the tools, and answer with nothing.""" - self.called = True self.tools = tools return ModelResponse(output=[], usage=Usage(), response_id=None) @@ -721,74 +721,59 @@ def stream_response( tracing: ModelTracing, **kwargs: Any, ) -> AsyncIterator[TResponseStreamEvent]: - """Record the tools, and stream nothing.""" - self.called = True self.tools = tools return _no_stream_events() def _hosted_mcp_config_the_model_received(model: _ToolRecordingModel) -> dict[str, Any]: - """The tool config of the single hosted MCP tool the model was handed.""" assert len(model.tools) == 1 tool = model.tools[0] assert isinstance(tool, HostedMCPTool) - return _fields(tool.tool_config) + return _as_dict(tool.tool_config) async def test_invoke_model_activity_resolves_tool_secrets( monkeypatch: pytest.MonkeyPatch, ): - """The model is handed the secret, not the marker the workflow sent.""" - monkeypatch.setenv(ENV_KEY, SENTINEL) - sent = _hosted_mcp_tool(secret_reference(ENV_KEY), "not-a-secret") + monkeypatch.setenv(ENV_NAME, SENTINEL) + sent = _hosted_mcp_tool(temporal_worker_env_ref(ENV_NAME), "not-a-secret") _payload, activity_input = _round_trip_activity_input(sent) model = _ToolRecordingModel() await ActivityEnvironment().run( - ModelActivity(TestModelProvider(model)).invoke_model_activity, + ModelActivity( + TestModelProvider(model), resolvable_worker_env_vars=[ENV_NAME] + ).invoke_model_activity, activity_input, ) assert _hosted_mcp_config_the_model_received(model) == { - **_fields(sent.tool_config), + **_as_dict(sent.tool_config), "authorization": SENTINEL, } -async def test_invoke_model_activity_rejects_a_malformed_domain_secret(): - """A rejection reaches the caller as the activity's own failure. - - A retryable escape would run the attempt again, with the same argument, - forever. - """ - _payload, activity_input = _round_trip_activity_input( - _code_interpreter_tool( - {"domain": "example.com", "name": "TOKEN", "vaule": SENTINEL} - ) - ) +async def test_invoke_model_activity_resolves_nothing_by_default( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv(ENV_NAME, SENTINEL) + sent = _hosted_mcp_tool(temporal_worker_env_ref(ENV_NAME), "not-a-secret") + _payload, activity_input = _round_trip_activity_input(sent) model = _ToolRecordingModel() - with pytest.raises(ApplicationError) as err: - await ActivityEnvironment().run( - ModelActivity(TestModelProvider(model)).invoke_model_activity, - activity_input, - ) + await ActivityEnvironment().run( + ModelActivity(TestModelProvider(model)).invoke_model_activity, + activity_input, + ) - assert err.value.non_retryable - assert err.value.type == "SecretReferenceFailure" - assert SENTINEL not in err.value.message - failure = _reported_failure(err.value) - assert not failure.HasField("cause") - assert SENTINEL.encode() not in failure.SerializeToString() - assert not model.called + assert _hosted_mcp_config_the_model_received(model) == _as_dict(sent.tool_config) async def test_invoke_model_activity_streaming_resolves_tool_secrets( monkeypatch: pytest.MonkeyPatch, client: Client ): - """The streaming activity hands the model the secret by its own path.""" - monkeypatch.setenv(ENV_KEY, SENTINEL) - sent = _hosted_mcp_tool(secret_reference(ENV_KEY), "not-a-secret") + monkeypatch.setenv(ENV_NAME, SENTINEL) + sent = _hosted_mcp_tool(temporal_worker_env_ref(ENV_NAME), "not-a-secret") _payload, activity_input = _round_trip_activity_input(sent) streaming_input: StreamingActivityModelInput = { **activity_input, @@ -797,44 +782,86 @@ async def test_invoke_model_activity_streaming_resolves_tool_secrets( model = _ToolRecordingModel() await ActivityEnvironment(client).run( - ModelActivity(TestModelProvider(model)).invoke_model_activity_streaming, + ModelActivity( + TestModelProvider(model), resolvable_worker_env_vars=[ENV_NAME] + ).invoke_model_activity_streaming, streaming_input, ) assert _hosted_mcp_config_the_model_received(model) == { - **_fields(sent.tool_config), + **_as_dict(sent.tool_config), "authorization": SENTINEL, } @workflow.defn -class SecretReferenceWorkflow: - """Builds a hosted MCP tool config the way a user's workflow would.""" - +class WorkerEnvRefAgentWorkflow: @workflow.run - async def run(self, key: str) -> str: + async def run(self) -> None: tool_config: Any = { "type": "mcp", "server_label": "test_server", "server_url": "https://example.com/mcp", - "authorization": secret_reference(key), + "authorization": temporal_worker_env_ref(ENV_NAME), + "headers": {"X-Token": temporal_worker_env_ref(OTHER_ENV_NAME)}, } - tool = HostedMCPTool(tool_config=tool_config) - return _fields(tool.tool_config)["authorization"] + agent = Agent[None]( + name="Worker env ref agent", + tools=[HostedMCPTool(tool_config=tool_config)], + ) + await Runner.run(starting_agent=agent, input="hi") -async def test_secret_reference_can_be_called_from_workflow_code(client: Client): - """A marker built in workflow code reaches the workflow result unchanged.""" +async def test_a_worker_resolves_only_the_variables_its_plugin_names( + monkeypatch: pytest.MonkeyPatch, client: Client +): + monkeypatch.setenv(ENV_NAME, SENTINEL) + monkeypatch.setenv(OTHER_ENV_NAME, OTHER_SENTINEL) + model = _ToolRecordingModel() + config = client.config() + config["plugins"] = [ + *config.get("plugins", []), + OpenAIAgentsPlugin( + model_provider=TestModelProvider(model), + resolvable_worker_env_vars=[ENV_NAME], + ), + ] + client = Client(**config) + + async with new_worker(client, WorkerEnvRefAgentWorkflow) as worker: + await client.execute_workflow( + WorkerEnvRefAgentWorkflow.run, + id=f"worker-env-ref-allowlist-{uuid.uuid4()}", + task_queue=worker.task_queue, + ) + + tool_config = _hosted_mcp_config_the_model_received(model) + assert tool_config["authorization"] == SENTINEL + assert tool_config["headers"] == { + "X-Token": temporal_worker_env_ref(OTHER_ENV_NAME) + } + + +async def test_an_agent_environment_forwards_the_variables_it_names_to_the_worker( + monkeypatch: pytest.MonkeyPatch, client: Client +): + monkeypatch.setenv(ENV_NAME, SENTINEL) + monkeypatch.setenv(OTHER_ENV_NAME, OTHER_SENTINEL) + model = _ToolRecordingModel() + async with AgentEnvironment( - model=TestModel.returning_responses([]), + model=model, resolvable_worker_env_vars=[ENV_NAME] ) as env: client = env.applied_on_client(client) - async with new_worker(client, SecretReferenceWorkflow) as worker: - result = await client.execute_workflow( - SecretReferenceWorkflow.run, - ENV_KEY, - id=f"secret-reference-workflow-{uuid.uuid4()}", + async with new_worker(client, WorkerEnvRefAgentWorkflow) as worker: + await client.execute_workflow( + WorkerEnvRefAgentWorkflow.run, + id=f"agent-environment-env-ref-allowlist-{uuid.uuid4()}", task_queue=worker.task_queue, ) - assert result == f"temporal.secret_reference:{ENV_KEY}" + tool_config = _hosted_mcp_config_the_model_received(model) + assert tool_config["authorization"] == SENTINEL + assert tool_config["headers"] == { + "X-Token": temporal_worker_env_ref(OTHER_ENV_NAME) + } From e8e5e245253cdb3a03b0e4bf3dd1802402a4fedd Mon Sep 17 00:00:00 2001 From: maplexu Date: Mon, 17 Aug 2026 13:48:34 -0400 Subject: [PATCH 5/5] AI-382: cut the hosted tool secrets docs to an introduction The README section had grown into a reference: which names resolve, what happens when one does not, what happens when an allowed name is unset, and where a reference is ignored. What is left is the purpose, one example, the worker's obligation, the matching rule, and the three fields substitution applies to. An unresolved reference reaches the MCP server and fails authentication, so those error paths document themselves. temporal_worker_env_ref's docstring loses the HostedMCPTool example it duplicated from the README, on a function whose signature is (str) -> str and whose return value the sentence beside it already places; the unset-resolves-empty fact stays there, since a caller can be surprised by it. The domain_secrets single-pass-iterator note moves from _resolve_network_policy's docstring onto the write-back it explains, where a maintainer about to delete an apparent self-assignment will read it. --- temporalio/contrib/openai_agents/README.md | 10 +++--- .../openai_agents/_temporal_openai_agents.py | 2 +- .../openai_agents/_temporal_worker_env_ref.py | 35 +++++-------------- 3 files changed, 14 insertions(+), 33 deletions(-) diff --git a/temporalio/contrib/openai_agents/README.md b/temporalio/contrib/openai_agents/README.md index f2a748100..89a9cc970 100644 --- a/temporalio/contrib/openai_agents/README.md +++ b/temporalio/contrib/openai_agents/README.md @@ -513,17 +513,15 @@ Every worker that runs model activities must both set `MY_MCP_TOKEN` and name it plugin = OpenAIAgentsPlugin(resolvable_worker_env_vars=["MY_MCP_TOKEN"]) ``` -Nothing resolves unless the worker names it. A reference to a name the worker does not allow is sent to the model provider as literal text. Names are matched exactly — no globbing, no prefixes — and `"*"` anywhere in the list allows every environment variable on the worker. +Names are matched exactly, with no globbing, and `"*"` anywhere in the list allows every environment variable on the worker. -A reference need not be the whole value: `"Bearer " + temporal_worker_env_ref("MY_MCP_TOKEN")` substitutes the credential inside a larger string. Each reference in a value is checked against the list on its own, so one value can end up holding both a resolved credential and the literal text of a name the worker does not allow. A name the worker does allow but has not set resolves to an empty value. +A reference can sit inside a larger value: in `"Bearer " + temporal_worker_env_ref("MY_MCP_TOKEN")`, the reference is replaced in place and the rest of the string is sent unchanged. -The variable's value is substituted in these fields and no others: +The environment variable's value is substituted in these fields and no others: - `authorization`, and the value of each entry in `headers`, in a `HostedMCPTool`'s `tool_config` - `value` in each entry of `network_policy.domain_secrets` under a hosted `ShellTool`'s `environment` -- `value` in each entry of `network_policy.domain_secrets` under a `CodeInterpreterTool`'s `container` - -Anywhere else — in a header *name*, or as an MCP server `factory_argument` (see [Factory Arguments](#factory-arguments)) — the reference is passed on as literal text, with no error from this SDK. +- `value` in each entry of `network_policy.domain_secrets` under the `container` in a `CodeInterpreterTool`'s `tool_config` ## Sandbox Support diff --git a/temporalio/contrib/openai_agents/_temporal_openai_agents.py b/temporalio/contrib/openai_agents/_temporal_openai_agents.py index d4d91130e..54beb9d75 100644 --- a/temporalio/contrib/openai_agents/_temporal_openai_agents.py +++ b/temporalio/contrib/openai_agents/_temporal_openai_agents.py @@ -249,7 +249,7 @@ def __init__( Use with caution in production environments. resolvable_worker_env_vars: Names of the environment variables that ``temporal_worker_env_ref()`` may read on this worker. Names are - matched exactly, with no globbing or prefix matching; ``"*"`` + matched exactly, with no globbing; ``"*"`` anywhere in the collection allows every name. Warning: resolvable_worker_env_vars is experimental and behavior may change in future versions. Use with caution in production environments. diff --git a/temporalio/contrib/openai_agents/_temporal_worker_env_ref.py b/temporalio/contrib/openai_agents/_temporal_worker_env_ref.py index 403c77314..503db3587 100644 --- a/temporalio/contrib/openai_agents/_temporal_worker_env_ref.py +++ b/temporalio/contrib/openai_agents/_temporal_worker_env_ref.py @@ -25,31 +25,15 @@ def temporal_worker_env_ref(name: str) -> str: Use with caution in production environments. Use it for a hosted tool credential that should come from the worker's - environment rather than being written into your workflow. Put the reference - returned in a ``HostedMCPTool``'s ``authorization`` or header value, or in the - ``value`` of a ``domain_secrets`` entry under a ``ShellTool`` or - ``CodeInterpreterTool``; the reference carries only the variable's name, never - its value. - - :: - - from agents import HostedMCPTool - - from temporalio.contrib.openai_agents import temporal_worker_env_ref - - tool = HostedMCPTool( - tool_config={ - "type": "mcp", - "server_label": "my_server", - "server_url": "https://example.com/mcp", - "authorization": temporal_worker_env_ref("MY_MCP_TOKEN"), - } - ) + environment rather than being written into your workflow. Put the returned + reference in a ``HostedMCPTool``'s ``authorization`` or header value, or in + the ``value`` of a ``domain_secrets`` entry under a ``ShellTool`` or + ``CodeInterpreterTool``. The reference carries only the variable's name, + never its value. Every worker that runs model activities must set the variable and name it in - ``OpenAIAgentsPlugin(resolvable_worker_env_vars=["MY_MCP_TOKEN"])``. A name the - worker does not allow is sent to the model provider as literal text; an allowed - name that is unset resolves to an empty value. + ``OpenAIAgentsPlugin(resolvable_worker_env_vars=[...])``. A name the worker + allows but has not set resolves to an empty value. Args: name: Name of the environment variable to read on the worker. @@ -83,14 +67,13 @@ def _resolve_domain_secret(self, secret: Mapping[str, Any]) -> dict[str, Any]: return {**secret, "value": self._resolve_ref(secret["value"])} def _resolve_network_policy(self, network_policy: Any) -> Any: - """On the code interpreter path pydantic deserializes ``domain_secrets`` into a - single-pass iterator, so the entries read here go back onto the input. - """ policy = cast(MutableMapping[str, Any], network_policy) domain_secrets = policy.get("domain_secrets") if domain_secrets is None: return network_policy unresolved = list(domain_secrets) + # On the code interpreter path pydantic deserializes domain_secrets into a + # single-pass iterator, so the entries read here go back onto the input. policy["domain_secrets"] = unresolved return { **policy,