Repository: pydantic/pydantic-ai
File: tests/test_mcp.py (Python)

Code before:
    assert output == snapshot(
        'The `pydantic/pydantic-ai` repository is a Python agent framework designed to facilitate the development of production-grade Generative AI applications and workflows with a focus on type-safety and an ergonomic developer experience.'
    )


async def test_custom_http_client_not_closed():
    custom_http_client = cached_async_http_client()

    assert not custom_http_client.is_closed

    my_mcp_server = MCPServerStreamableHTTP(
        url='https://mcp.deepwiki.com/mcp', http_client=custom_http_client, timeout=30
    )

    tools = await my_mcp_server.list_tools()
    assert len(tools) > 0

    assert not custom_http_client.is_closed


# ============================================================================
# Tool and Resource Caching Tests
# ============================================================================


async def test_tools_caching_enabled_by_default(mcp_server: MCPServerStdio) -> None:
    """Test that list_tools() caches results by default."""
    async with mcp_server:
        # First call - should fetch from server and cache
        tools1 = await mcp_server.list_tools()
        assert len(tools1) > 0
        assert mcp_server._cached_tools is not None  # pyright: ignore[reportPrivateUsage]

        # Mock _client.list_tools to track if it's called again
        original_list_tools = mcp_server._client.list_tools  # pyright: ignore[reportPrivateUsage]
        call_count = 0

        async def mock_list_tools():  # pragma: no cover
            nonlocal call_count
            call_count += 1
            return await original_list_tools()

        mcp_server._client.list_tools = mock_list_tools  # pyright: ignore[reportPrivateUsage,reportAttributeAccessIssue]

        # Second call - should return cached value without calling server
        tools2 = await mcp_server.list_tools()
        assert tools2 == tools1
        assert call_count == 0


async def test_tools_no_caching_when_disabled() -> None:

Review comment (suggestion):
Instead of mocking, can we test against a real MCP server, specifically `tests.mcp_server` like the rest of this file? You could give it a tool that is hidden by default, plus a special `enable_hidden_tool` tool that will cause the tool to be able and the notification to be sent.

I assume FastMCP 1.0, which we use in that `tests.mcp_server`, supports something like that

Code after:
        assert mcp_server.capabilities.resources is True
        assert mcp_server.capabilities.tools is True
        assert mcp_server.capabilities.prompts is True
        assert mcp_server.capabilities.logging is True
        assert mcp_server.capabilities.completions is False
        assert mcp_server.capabilities.experimental is None


async def test_resource_methods_without_capability(mcp_server: MCPServerStdio) -> None:
    """Test that resource list methods return empty values when resources capability is not available."""
    async with mcp_server:
        # Mock the capabilities to not support resources
        mock_capabilities = ServerCapabilities(resources=False)
        with patch.object(mcp_server, '_server_capabilities', mock_capabilities):
            # list_resources should return empty list
            result = await mcp_server.list_resources()
            assert result == []

            # list_resource_templates should return empty list
            result = await mcp_server.list_resource_templates()
            assert result == []


async def test_instructions(mcp_server: MCPServerStdio) -> None:
    with pytest.raises(
        AttributeError, match='The `MCPServerStdio.instructions` is only available after initialization.'
    ):
        mcp_server.instructions
    async with mcp_server:
        assert mcp_server.instructions == 'Be a helpful assistant.'


async def test_agent_run_stream_with_mcp_server_http(allow_model_requests: None, model: Model):
    server = MCPServerStreamableHTTP(url='https://mcp.deepwiki.com/mcp', timeout=30)
    agent = Agent(model, toolsets=[server], instructions='Be concise.')

    # This should not raise an error.
    # See https://github.com/pydantic/pydantic-ai/issues/2818#issuecomment-3476480829
    async with agent.run_stream('Summarize the pydantic/pydantic-ai repo in one sentence') as result:
        output = await result.get_output()
    assert output == snapshot(
        'The `pydantic/pydantic-ai` repository is a Python agent framework designed to facilitate the development of production-grade Generative AI applications and workflows with a focus on type-safety and an ergonomic developer experience.'
    )


async def test_custom_http_client_not_closed():
    custom_http_client = cached_async_http_client()

    assert not custom_http_client.is_closed

    my_mcp_server = MCPServerStreamableHTTP(