MCP Client example: Chatbot CLI

client.py

import asyncio
from typing import Optional
from contextlib import AsyncExitStack

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()  # load environment variables from .env

class MCPClient:
    def __init__(self):
        # Initialize session and client objects
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self.anthropic = Anthropic()

    async def connect_to_server(self, server_script_path: str):
        """Connect to an MCP server
        
        Args:
            server_script_path: Path to the server script (.py or .js)
        """
        is_python = server_script_path.endswith('.py')
        is_js = server_script_path.endswith('.js')
        if not (is_python or is_js):
            raise ValueError("Server script must be a .py or .js file")
            
        command = "python" if is_python else "node"
        server_params = StdioServerParameters(
            command=command,
            args=[server_script_path],
            env=None
        )
        
        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
        self.stdio, self.write = stdio_transport
        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
        
        await self.session.initialize()
        
        # List available tools
        response = await self.session.list_tools()
        tools = response.tools
        print("\nConnected to server with tools:", [tool.name for tool in tools])

    async def process_query(self, query: str) -> str:
        """Process a query using Claude and available tools"""
        messages = [
            {
                "role": "user",
                "content": query
            }
        ]

        response = await self.session.list_tools()
        available_tools = [{ 
            "name": tool.name,
            "description": tool.description,
            "input_schema": tool.inputSchema
        } for tool in response.tools]

        # Initial Claude API call
        response = self.anthropic.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1000,
            messages=messages,
            tools=available_tools
        )

        # Process response and handle tool calls
        tool_results = []
        final_text = []

        for content in response.content:
            if content.type == 'text':
                final_text.append(content.text)
            elif content.type == 'tool_use':
                tool_name = content.name
                tool_args = content.input
                
                # Execute tool call
                result = await self.session.call_tool(tool_name, tool_args)
                tool_results.append({"call": tool_name, "result": result})
                final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")

                # Continue conversation with tool results
                if hasattr(content, 'text') and content.text:
                    messages.append({
                      "role": "assistant",
                      "content": content.text
                    })
                messages.append({
                    "role": "user", 
                    "content": result.content
                })

                # Get next response from Claude
                response = self.anthropic.messages.create(
                    model="claude-3-5-sonnet-20241022",
                    max_tokens=1000,
                    messages=messages,
                )

                final_text.append(response.content[0].text)

        return "\n".join(final_text)

    async def chat_loop(self):
        """Run an interactive chat loop"""
        print("\nMCP Client Started!")
        print("Type your queries or 'quit' to exit.")
        
        while True:
            try:
                query = input("\nQuery: ").strip()
                
                if query.lower() == 'quit':
                    break
                    
                response = await self.process_query(query)
                print("\n" + response)
                    
            except Exception as e:
                print(f"\nError: {str(e)}")
    
    async def cleanup(self):
        """Clean up resources"""
        await self.exit_stack.aclose()

async def main():
    if len(sys.argv) < 2:
        print("Usage: python client.py <path_to_server_script>")
        sys.exit(1)
        
    client = MCPClient()
    try:
        await client.connect_to_server(sys.argv[1])
        await client.chat_loop()
    finally:
        await client.cleanup()

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

[WINDOWS] If you’re continuing the weather.py tutorial from the server quickstart and the 2 projects are on different folders like /mcp-client and /weather, you'll have to connect the weather.py file to the /mcp-client .venv, or the python interpreter won't find the lib.

Error message:

(mcp-client) PS C:\Users\Robin Roy\Desktop\mcp-client> uv run client.py 'C:/Users/Robin Roy/Desktop/learnmcp/weather.py'
Traceback (most recent call last):
  File "C:\Users\Robin Roy\Desktop\learnmcp\weather.py", line 3, in <module>
    from mcp.server.fastmcp import FastMCP
ModuleNotFoundError: No module named 'mcp'

Fix:
Make sure to activate the .venv of mcp-client (and have the httpx library added using uv add httpx)
Activate venv: .venv/Scripts/activate

        command = "python" if is_python else "node"
        # Create a modified environment to include virtual environment path
        env = os.environ.copy()
        venv_site_packages = os.path.join(sys.prefix, 'Lib', 'site-packages')
        if 'PYTHONPATH' in env:
            env['PYTHONPATH'] = f"{venv_site_packages}:{env['PYTHONPATH']}"
        else:
            env['PYTHONPATH'] = venv_site_packages

        server_params = StdioServerParameters(
            command=command,
            args=[server_script_path],
            env=env
        )

To run, the command remains the same uv run client.py 'path/to/weather.py'

Hi, thank you so much for providing such an excellent example code—it has clarified most of my questions. However, I’m a bit puzzled by one aspect: why is the LLM invocation happening on the client side? For instance, in the process_query function, it concatenates all the tools from the current MCP Server and sends them to the Claude API. If there are additional MCP servers, wouldn’t this approach potentially overlook the definitions of their tools?
I appreciate your time and assistance!

For line 95-97

               messages.append({
                    "role": "user", 
                    "content": result.content
                })

Shouldn't the content be a tool result and the message should also include the model's tool calling like on the website (see below)?

             messages.append({
                "role": "assistant",
                "content": assistant_message_content
            })
            messages.append({
                "role": "user",
                "content": [
                    {
                        "type": "tool_result",
                        "tool_use_id": content.id,
                        "content": result.content
                    }
                ]
            })

Hi is there any resource to interface this setup with gemini model

Hi is there any resource to interface this setup with gemini model

@2137942shubham
this is a way in which the interface can be setup with gemini

async def connect_to_server(self, server_script_path: str):
        """Connect to an MCP server

        Args:
            server_script_path: Path to the server script (.py or .js)
        """
        is_python = server_script_path.endswith('.py')
        is_js = server_script_path.endswith('.js')
        if not (is_python or is_js):
            raise ValueError("Server script should be a .py or .js file")
        
        command = "python" if is_python else "node"
        server_params = StdioServerParameters(
            command=command,
            args=[server_script_path],
            env = None
        )

        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
        self.stdio, self.write = stdio_transport
        self.sessions = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))

        await self.sessions.initialize()

        # List available tools
        response = await self.sessions.list_tools()
        tools = response.tools
        print("\nConnected to server with tools:",[tool.name for tool in tools])

        available_tools = []
        for tool in tools:
            tool_definition = {
                "function_declarations": [
                    {
                        "name": tool.name,
                        "description": tool.description
                    }
                ]
            }
            available_tools.append(tool_definition)

        self.model = genai.GenerativeModel(
            model_name="gemini-2.0-flash",
            tools=available_tools
        )

        self.chat = self.model.start_chat()

Hi there, can anyone help me with

import asyncio
from typing import Optional
from contextlib import AsyncExitStack
import logging
import json

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()  # load environment variables from .env
print("Environment variables loaded from .env file")

class MCPClient:
    def __init__(self):
        # Initialize session and client objects
        print("Initializing MCPClient...")
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self.anthropic = Anthropic()
        print("MCPClient initialized successfully")

    async def connect_to_server(self, server_script_path: str):
        """Connect to an MCP server
        
        Args:
            server_script_path: Path to the server script (.py or .js)
        """
        print(f"Connecting to server with script: {server_script_path}")
        is_python = server_script_path.endswith('.py')
        is_js = server_script_path.endswith('.js')
        if not (is_python or is_js):
            print(f"Error: Invalid script type for {server_script_path}")
            raise ValueError("Server script must be a .py or .js file")
            
        command = "python" if is_python else "node"
        print(f"Using {command} to execute server script")
        server_params = StdioServerParameters(
            command=command,
            args=[server_script_path],
            env=None
        )
        
        print("Establishing stdio transport connection...")
        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
        self.stdio, self.write = stdio_transport
        print("Creating client session...")
        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
        
        print("Initializing session...")
        await self.session.initialize()
        
        # List available tools
        print("Retrieving available tools...")
        response = await self.session.list_tools()
        tools = response.tools
        print("\nConnected to server with tools:", [tool.name for tool in tools])
        print(f"Total tools available: {len(tools)}")```
        
        
        I have this code but it is getting strucked at Initialing session forever. It is not showing any error or anything

Hi there, can anyone help me with

import asyncio
from typing import Optional
from contextlib import AsyncExitStack
import logging
import json

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()  # load environment variables from .env
print("Environment variables loaded from .env file")

class MCPClient:
    def __init__(self):
        # Initialize session and client objects
        print("Initializing MCPClient...")
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self.anthropic = Anthropic()
        print("MCPClient initialized successfully")

    async def connect_to_server(self, server_script_path: str):
        """Connect to an MCP server
        
        Args:
            server_script_path: Path to the server script (.py or .js)
        """
        print(f"Connecting to server with script: {server_script_path}")
        is_python = server_script_path.endswith('.py')
        is_js = server_script_path.endswith('.js')
        if not (is_python or is_js):
            print(f"Error: Invalid script type for {server_script_path}")
            raise ValueError("Server script must be a .py or .js file")
            
        command = "python" if is_python else "node"
        print(f"Using {command} to execute server script")
        server_params = StdioServerParameters(
            command=command,
            args=[server_script_path],
            env=None
        )
        
        print("Establishing stdio transport connection...")
        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
        self.stdio, self.write = stdio_transport
        print("Creating client session...")
        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
        
        print("Initializing session...")
        await self.session.initialize()
        
        # List available tools
        print("Retrieving available tools...")
        response = await self.session.list_tools()
        tools = response.tools
        print("\nConnected to server with tools:", [tool.name for tool in tools])
        print(f"Total tools available: {len(tools)}")```
        
        
        I have this code but it is getting strucked at Initialing session forever. It is not showing any error or anything

Can you tell where you have initialised the Claude tool system as in this self.anthropic = Anthropic() only instantiates the anthropic client.

Hi there, can anyone help me with

import asyncio
from typing import Optional
from contextlib import AsyncExitStack
import logging
import json

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()  # load environment variables from .env
print("Environment variables loaded from .env file")

class MCPClient:
    def __init__(self):
        # Initialize session and client objects
        print("Initializing MCPClient...")
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self.anthropic = Anthropic()
        print("MCPClient initialized successfully")

    async def connect_to_server(self, server_script_path: str):
        """Connect to an MCP server
        
        Args:
            server_script_path: Path to the server script (.py or .js)
        """
        print(f"Connecting to server with script: {server_script_path}")
        is_python = server_script_path.endswith('.py')
        is_js = server_script_path.endswith('.js')
        if not (is_python or is_js):
            print(f"Error: Invalid script type for {server_script_path}")
            raise ValueError("Server script must be a .py or .js file")
            
        command = "python" if is_python else "node"
        print(f"Using {command} to execute server script")
        server_params = StdioServerParameters(
            command=command,
            args=[server_script_path],
            env=None
        )
        
        print("Establishing stdio transport connection...")
        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
        self.stdio, self.write = stdio_transport
        print("Creating client session...")
        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
        
        print("Initializing session...")
        await self.session.initialize()
        
        # List available tools
        print("Retrieving available tools...")
        response = await self.session.list_tools()
        tools = response.tools
        print("\nConnected to server with tools:", [tool.name for tool in tools])
        print(f"Total tools available: {len(tools)}")```
        
        
        I have this code but it is getting strucked at Initialing session forever. It is not showing any error or anything

Does your server.py file throw any error when you run "uv run server.py" command?

Regarding @robinroy03 comment for server env, I have come to another solution when both client and server use a virtual env
This time I assume that the server python file is at the project root folder
so, what we have to set at "env" parameter of StdioServerParameters it is server env, not client env.
That's way we have to switch the folder name in the VIRTUAL_ENV var, because the clone will have the client dir, not server's. And also the path must include server's vent's bin dir

        server_env = os.environ.copy()
        # Set the server's virtual environment path
        server_venv_path = os.path.join(os.path.dirname(server_script_path), '.venv')
        server_env['VIRTUAL_ENV'] = server_venv_path

        # Update PATH to include the server's virtual environment executables
        server_env['PATH'] = os.path.join(server_venv_path, 'bin') + os.pathsep + server_env['PATH']

        server_params = StdioServerParameters(
            command=command,
            args=[server_script_path],
            env=server_env
        )

In my case when ask the weather of some location and alert of some other, it says that its calling functions but then dosn't

We need to purchase the Claude API key right?

We need to purchase the Claude API key right?

yup

can any tell me whether it is importnat to use the uv , can we just simply use pip to install the packages andd creatting the virtual environment
because pip is widely being used to make agentic applications, and one is familiar to it, can we like use it

When I ran the code successfully, I found that the client did not invoke the server's tools. Have you ever encountered such a situation?
`Connected to server with tools: ['get_alerts', 'get_forecast']

MCP Client Started!
Type your queries or 'quit' to exit.

Query: What are the weather alerts in California

I apologize, but I am not able to provide real-time weather alert information for California. As an AI coding assistant, I don't have access to live weather data or alert systems.

To get accurate weather alerts for California, I recommend:

1. Visiting the National Weather Service website (weather.gov) and searching for California
2. Checking your local news station's weather reports
3. Using the official NOAA Weather app
4. Setting up alerts through the FEMA app
5. Following your local emergency management office on social media

These sources will provide you with current, accurate weather alerts and emergency information for your specific location in California.

Would you like help writing code to integrate weather alert data into an application instead? I'd be happy to assist with that type of programming task.
`

We need to purchase the Claude API key right?

yup

Can we use another API Key from Genmini Google AI Studio? because it's free

help me fix this error, when I run the client .. AttributeError: 'MCPClient' object has no attribute 'connect_to_server'

We need to purchase the Claude API key right?

yup

Can we use another API Key from Genmini Google AI Studio? because it's free

Yes, we can

We need to purchase the Claude API key right?

yup

Can we use another API Key from Genmini Google AI Studio? because it's free

Yes, we can

oki, Got it thank u