Agent with MCP in google Colab

[dominicdev]

Initial Checks

• I confirm that I'm using the latest version of Pydantic AI
• I confirm that I searched for my issue in https://github.com/pydantic/pydantic-ai/issues before opening this issue

Description

I'm trying to test a MCP in google colab , using the basic code in the docs

but getting fileno error in part of MCPServerStdio

errro code in google colab

UnsupportedOperation                      Traceback (most recent call last)
[/tmp/ipython-input-3697815817.py](https://localhost:8080/#) in <cell line: 0>()
     45 
     46 if __name__ == "__main__":
---> 47     asyncio.run(main())

25 frames
[/usr/local/lib/python3.12/dist-packages/ipykernel/iostream.py](https://localhost:8080/#) in fileno(self)
    309             return self._original_stdstream_copy
    310         else:
--> 311             raise io.UnsupportedOperation("fileno")
    312 
    313     def _watch_pipe_fd(self):

UnsupportedOperation: fileno

Example Code

here is example code


import asyncio
from typing import Any
from pydantic_ai import Agent
from pydantic_ai.mcp import CallToolFunc, MCPServerStdio, ToolResult
from pydantic_ai.models.test import TestModel
from pydantic_ai.tools import RunContext
from pydantic_ai.models.google import GoogleModel
from pydantic_ai.providers.google import GoogleProvider
import os
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
provider = GoogleProvider(api_key=os.environ["GOOGLE_API_KEY"])
model = GoogleModel('gemini-1.5-flash', provider=provider)

PROMPT = 'Summarize this video into the key sections, ideas and action items. Deliver the notes from a first person perspective. If there are slides or info graphics in the video then include a description of them with any text that appears on there. Also make a transcript from the video after summarize sa it to md file,the filename will be date today in DDMMYYYY format'

import nest_asyncio
nest_asyncio.apply()
 
server = MCPServerStdio(command='python', args=['/content/yt_transcript.py'],)
print("server",server)
agent = Agent(model, toolsets=[server])



async def main():
    async with agent:
        agent.set_mcp_sampling_model()
        result = await agent.run('Create an image of a robot in a punk style.')
        print(result.output)

Python, Pydantic AI & LLM client version

Python 3.12.11

[DouweM]

This is an upstream issue in the mcp library: modelcontextprotocol/python-sdk#1103, caused by the default errlog = sys.stderr argument to stdio_client here:

https://github.com/modelcontextprotocol/python-sdk/blob/07ae8c0d4e58bc0527a6898709288686ef6506dd/src/mcp/client/stdio/__init__.py#L106

As a workaround, you can try defining a custom subclass of MCPServerStdio that passes errlog=None:

class MCPServerStdioNoErrlog(MCPServerStdio):
    @asynccontextmanager
    async def client_streams(
        self,
    ) -> AsyncIterator[
        tuple[
            MemoryObjectReceiveStream[SessionMessage | Exception],
            MemoryObjectSendStream[SessionMessage],
        ]
    ]:
        server = StdioServerParameters(command=self.command, args=list(self.args), env=self.env, cwd=self.cwd)
        async with stdio_client(server=server, errlog=None) as (read_stream, write_stream):
            yield read_stream, write_stream

Let me know if that works. If so, we can add a field for that argument to MCPServerStdio.

[dominicdev]

This is an upstream issue in the mcp library: modelcontextprotocol/python-sdk#1103, caused by the default errlog = sys.stderr argument to stdio_client here:

https://github.com/modelcontextprotocol/python-sdk/blob/07ae8c0d4e58bc0527a6898709288686ef6506dd/src/mcp/client/stdio/__init__.py#L106

As a workaround, you can try defining a custom subclass of MCPServerStdio that passes errlog=None:

class MCPServerStdioNoErrlog(MCPServerStdio):
@asynccontextmanager
async def client_streams(
self,
) -> AsyncIterator[
tuple[
MemoryObjectReceiveStream[SessionMessage | Exception],
MemoryObjectSendStream[SessionMessage],
]
]:
server = StdioServerParameters(command=self.command, args=list(self.args), env=self.env, cwd=self.cwd)
async with stdio_client(server=server, errlog=None) as (read_stream, write_stream):
yield read_stream, write_stream
Let me know if that works. If so, we can add a field for that argument to MCPServerStdio.
worked!!

but also change the mcp to sse and got no problems