[Bug] openai_agents: tool guardrails on the MCP factory's server are silently skipped #303

"""Tool guardrails on the MCP factory's server are silently skipped.

python issue_repro.py    (starts a local Temporal dev server; no API key needed)

With openai-agents 0.22.3, an MCP server takes tool_input_guardrails. A guardrail
that rejects every `wire` call stops the tool in a native run, and
temporal_mcp_server(tool_input_guardrails=...) does the same in a Workflow. Set
on the server that the Worker's factory returns, the guardrail is skipped on all
three routes: the tool runs and nothing fails or warns. Exits 1 while a
guardrail on the factory's server lets the tool run.
"""

import asyncio
import sys
import uuid
from datetime import timedelta
from typing import Any

from temporalio import workflow
from temporalio.client import Client
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker

with workflow.unsafe.imports_passed_through():
    from agents import (
        Agent,
        Model,
        RunConfig,
        Runner,
        ToolGuardrailFunctionOutput,
        ToolInputGuardrailData,
        tool_input_guardrail,
    )
    from agents.mcp import MCPServerStdio
    from temporalio.openai_agents import (
        OpenAIAgentsPlugin,
        StatefulMCPServerProvider,
        StatelessMCPServerProvider,
    )
    from temporalio.openai_agents.testing import ResponseBuilders, TestModelProvider
    from temporalio.openai_agents.workflow import (
        stateful_mcp_server,
        stateless_mcp_server,
        temporal_mcp_server,
    )

# The MCP server, run over stdio with `python -c`: one tool, `wire`.
SERVER = """
from mcp.server.mcpserver import MCPServer

app = MCPServer("bank")


@app.tool()
def wire(amount: int) -> str:
    return "wired"


app.run("stdio")
"""


@tool_input_guardrail
def block_wire(data: ToolInputGuardrailData) -> ToolGuardrailFunctionOutput:
    return ToolGuardrailFunctionOutput.reject_content("blocked by guardrail")


def bank_server(**options: Any) -> MCPServerStdio:
    return MCPServerStdio(params={"command": sys.executable, "args": ["-c", SERVER]}, name="bank", **options)


def guarded_server() -> MCPServerStdio:
    return bank_server(tool_input_guardrails=[block_wire])


class WireModel(Model):
    """Calls `wire`, then answers with what the tool returned."""

    async def get_response(self, system_instructions, input, model_settings, tools, *args, **kwargs):
        items = [] if isinstance(input, str) else input
        outputs = [i for i in items if isinstance(i, dict) and i.get("type") == "function_call_output"]
        if outputs:
            return ResponseBuilders.output_message(str(outputs[-1]["output"]))
        return ResponseBuilders.tool_call('{"amount": 5}', "wire")

    def stream_response(self, *args, **kwargs):
        raise NotImplementedError


@workflow.defn
class WireWorkflow:
    @workflow.run
    async def run(self, case: str) -> str:
        if case == "temporal_mcp_server(tool_input_guardrails=)":
            server = temporal_mcp_server("bank", tool_input_guardrails=[block_wire])
        elif case == "temporal_mcp_server(), guardrail on factory":
            server = temporal_mcp_server("bank-guarded")
        elif case == "stateless_mcp_server(), guardrail on factory":
            server = stateless_mcp_server("bank-guarded")
        else:
            async with stateful_mcp_server("bank-guarded") as stateful:
                result = await Runner.run(Agent(name="bank", mcp_servers=[stateful]), "Wire 5.")
            return result.final_output
        result = await Runner.run(Agent(name="bank", mcp_servers=[server]), "Wire 5.")
        return result.final_output


async def main() -> int:
    async with guarded_server() as server:
        result = await Runner.run(Agent(name="bank", mcp_servers=[server]), "Wire 5.",
                                  run_config=RunConfig(model_provider=TestModelProvider(WireModel())))
    print(f"{'native MCPServerStdio(tool_input_guardrails=)':47} tool output: {result.final_output}")

    plugin = OpenAIAgentsPlugin(
        model_provider=TestModelProvider(WireModel()),
        mcp_servers={"bank": bank_server, "bank-guarded": guarded_server},
        mcp_server_providers=[
            StatelessMCPServerProvider("bank-guarded", guarded_server),
            StatefulMCPServerProvider("bank-guarded", lambda _: guarded_server()),
        ],
    )
    unguarded = 0
    async with await WorkflowEnvironment.start_local() as env:
        client = Client(**{**env.client.config(), "plugins": [plugin]})
        async with Worker(client, task_queue="bank", workflows=[WireWorkflow]):
            for case in ("temporal_mcp_server(tool_input_guardrails=)",
                         "temporal_mcp_server(), guardrail on factory",
                         "stateless_mcp_server(), guardrail on factory",
                         "stateful_mcp_server(), guardrail on factory"):
                output = await client.execute_workflow(
                    WireWorkflow.run, case, id=f"wire-{uuid.uuid4()}", task_queue="bank",
                    execution_timeout=timedelta(seconds=60))
                if "on factory" in case and "wired" in output:
                    unguarded += 1
                print(f"{case:47} tool output: {output}")
    return 1 if unguarded else 0


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))