Repository: pydantic/pydantic-ai
File: pydantic_ai_slim/pydantic_ai/mcp.py (Python)

Code before:
    def instructions(self) -> str | None:
        """Access the instructions sent by the MCP server during initialization."""
        if not hasattr(self, '_instructions'):
            raise AttributeError(
                f'The `{self.__class__.__name__}.instructions` is only available after initialization.'
            )
        return self._instructions

    async def list_tools(self) -> list[mcp_types.Tool]:
        """Retrieve tools that are currently active on the server.

        Tools are cached when the server advertises `tools.listChanged` capability,
        with cache invalidation on tool change notifications and reconnection.
        """
        async with self:  # Ensure server is running
            # Only cache if server supports listChanged notifications
            if self._server_capabilities.tools_list_changed:
                if self._cached_tools is not None and self._tools_cache_valid:
                    return self._cached_tools

                result = await self._client.list_tools()
                self._cached_tools = result.tools
                self._tools_cache_valid = True
                return result.tools
            else:
                # Server doesn't support notifications, always fetch fresh
                result = await self._client.list_tools()
                return result.tools

    async def direct_call_tool(
        self,
        name: str,
        args: dict[str, Any],
        metadata: dict[str, Any] | None = None,
    ) -> ToolResult:
        """Call a tool on the server.

        Args:
            name: The name of the tool to call.
            args: The arguments to pass to the tool.
            metadata: Request-level metadata (optional)

        Returns:
            The result of the tool call.

        Raises:
            ModelRetry: If the tool call fails.
        """
        async with self:  # Ensure server is running
            try:
                result = await self._client.send_request(

Review comment (bug):
I think typically, if a server doesn't support notifications, that means the tools will never change, so we should cache.

Although @phemmer said on the other PR, "a flag might be good for broken mcp servers which change tools without notifications", which I agree with: `cache_tools` set to `True` by default, and I suppose `cache_resources` as well.

Then we should always read/write the cache unless that flag is disabled, with no branching on the server capability.

Code after:
        else:
            return repr(self)

    @property
    def tool_name_conflict_hint(self) -> str:
        return 'Set the `tool_prefix` attribute to avoid name conflicts.'

    @property
    def server_info(self) -> mcp_types.Implementation:
        """Access the information send by the MCP server during initialization."""
        if getattr(self, '_server_info', None) is None:
            raise AttributeError(
                f'The `{self.__class__.__name__}.server_info` is only instantiated after initialization.'
            )
        return self._server_info

    @property
    def capabilities(self) -> ServerCapabilities:
        """Access the capabilities advertised by the MCP server during initialization."""
        if getattr(self, '_server_capabilities', None) is None:
            raise AttributeError(
                f'The `{self.__class__.__name__}.capabilities` is only instantiated after initialization.'
            )
        return self._server_capabilities

    @property
    def instructions(self) -> str | None:
        """Access the instructions sent by the MCP server during initialization."""
        if not hasattr(self, '_instructions'):
            raise AttributeError(
                f'The `{self.__class__.__name__}.instructions` is only available after initialization.'
            )
        return self._instructions

    async def list_tools(self) -> list[mcp_types.Tool]:
        """Retrieve tools that are currently active on the server.

        Tools are cached by default, with cache invalidation on:
        - `notifications/tools/list_changed` notifications from the server
        - Connection close (cache is cleared in `__aexit__`)

        Set `cache_tools=False` for servers that change tools without sending notifications.
        """
        async with self:
            if self.cache_tools:
                if self._cached_tools is not None:
                    return self._cached_tools
                result = await self._client.list_tools()
                self._cached_tools = result.tools
                return result.tools
            else: