[Bug] openai_agents: a callable tool_filter on a deprecated MCP provider's server makes tool listing retry forever #290

"""A callable tool_filter on a deprecated MCP provider's server makes every tool listing fail.

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

A callable tool_filter hides the `danger` tool from a native run, and
temporal_mcp_server(tool_filter=...) does the same in a Workflow. Set on the
server that the Worker's factory returns, it gets the run context and agent only
in the Workflow. temporal_mcp_server() rejects it at once with a clear error.
StatelessMCPServerProvider and StatefulMCPServerProvider call list_tools() with
no context, the Agents SDK raises UserError, and the list Activity retries it
forever: the run never reaches the model. Exits 1 while such a run is still
retrying after 30 seconds.
"""

import asyncio
import sys
import uuid
from typing import Any

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

with workflow.unsafe.imports_passed_through():
    from agents import Agent, Model, RunConfig, Runner
    from agents.mcp import MCPServerStdio, ToolFilterContext
    from mcp import Tool as MCPTool
    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,
    )

OFFERED: list[list[str]] = []  # the tool names the model is offered, per model call

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

app = MCPServer("tools")


@app.tool()
def safe() -> str:
    return "safe ran"


@app.tool()
def danger() -> str:
    return "danger ran"


app.run("stdio")
"""


def drop_danger(context: ToolFilterContext, tool: MCPTool) -> bool:
    """The documented dynamic filter shape: it reads the agent from the context."""
    return not (context.agent.name == "helper" and tool.name == "danger")


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


def filtered_server() -> MCPServerStdio:
    return stdio_server(tool_filter=drop_danger)


class SafeModel(Model):
    """Calls `safe` if it is offered, then answers with the tool's output."""

    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(f"done: {outputs[-1]['output']}")
        OFFERED.append(sorted(t.name for t in tools))
        return ResponseBuilders.tool_call("{}", "safe")

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


@workflow.defn
class ToolsWorkflow:
    @workflow.run
    async def run(self, case: str) -> str:
        if case == "temporal_mcp_server(tool_filter=)":
            server = temporal_mcp_server("tools", tool_filter=drop_danger)
        elif case == "temporal_mcp_server(), filter on factory":
            server = temporal_mcp_server("tools-filtered")
        elif case == "stateless_mcp_server(), filter on factory":
            server = stateless_mcp_server("tools-filtered")
        else:
            async with stateful_mcp_server("tools-filtered") as stateful:
                result = await Runner.run(Agent(name="helper", mcp_servers=[stateful]), "Go.")
            return result.final_output
        result = await Runner.run(Agent(name="helper", mcp_servers=[server]), "Go.")
        return result.final_output


async def main() -> int:
    async with filtered_server() as server:
        result = await Runner.run(Agent(name="helper", mcp_servers=[server]), "Go.",
                                  run_config=RunConfig(model_provider=TestModelProvider(SafeModel())))
    print(f"{'native MCPServerStdio(tool_filter=)':45} offered={OFFERED[-1]} output={result.final_output!r}")

    plugin = OpenAIAgentsPlugin(
        model_provider=TestModelProvider(SafeModel()),
        mcp_servers={"tools": stdio_server, "tools-filtered": filtered_server},
        mcp_server_providers=[
            StatelessMCPServerProvider("tools-filtered", filtered_server),
            StatefulMCPServerProvider("tools-filtered", lambda _: filtered_server()),
        ],
    )
    retrying = 0
    async with await WorkflowEnvironment.start_local() as env:
        client = Client(**{**env.client.config(), "plugins": [plugin]})
        async with Worker(client, task_queue="tools", workflows=[ToolsWorkflow]):
            for case in ("temporal_mcp_server(tool_filter=)",
                         "temporal_mcp_server(), filter on factory",
                         "stateless_mcp_server(), filter on factory",
                         "stateful_mcp_server(), filter on factory"):
                calls = len(OFFERED)
                handle = await client.start_workflow(
                    ToolsWorkflow.run, case, id=f"tools-{uuid.uuid4()}", task_queue="tools")
                try:
                    output = await asyncio.wait_for(handle.result(), timeout=30)
                    outcome = f"offered={OFFERED[-1]} output={output!r}"
                except WorkflowFailureError as err:
                    cause: BaseException = err
                    while getattr(cause, "cause", None) is not None:
                        cause = cause.cause  # type: ignore[attr-defined]
                    outcome = f"failed: {cause}"
                except asyncio.TimeoutError:
                    pending = (await handle.describe()).raw_description.pending_activities
                    await handle.terminate("still retrying after 30 seconds")
                    retrying += 1
                    outcome = "still running after 30 s: " + "; ".join(
                        f"{p.activity_type.name} attempt {p.attempt}, last failure "
                        f"{p.last_failure.application_failure_info.type}: {p.last_failure.message}"
                        for p in pending)
                print(f"{case:45} {outcome} model_calls={len(OFFERED) - calls}")
    return 1 if retrying else 0


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