Model Context Protocol (MCP)

$
# Create project directory
>
mkdir mcp-client
>
cd mcp-client
>
>
# Create virtual environment
>
python -m venv .venv
>
>
# Activate virtual environment
>
# On Windows:
>
.venv\Scripts\activate
>
# On Unix or MacOS:
>
source .venv/bin/activate
>
>
# Install required packages
>
pip install mcp uv anthropic python-dotenv
>
>
# Create our main file
>
touch client.py

$
# Add your ANTHROPIC_API_KEY and MERGE_API_KEY to .env
>
echo "ANTHROPIC_API_KEY=<your Anthropic key here>" >> .env
>
echo "MERGE_API_KEY=<your Merge key here>" >> .env
>
>
# Add .env file to .gitignore
>
echo ".env" >> .gitignore

1
import os
2
import asyncio
3
from typing import Optional
4
from contextlib import AsyncExitStack
5
6
from mcp import ClientSession, StdioServerParameters
7
from mcp.client.stdio import stdio_client
8
9
from anthropic import Anthropic
10
from dotenv import load_dotenv
11
12
load_dotenv()  # load environment variables from .env
13
14
class MCPClient:
15
    def __init__(self):
16
        # Initialize session and client objects
17
        self.session: Optional[ClientSession] = None
18
        self.exit_stack = AsyncExitStack()
19
        self.anthropic = Anthropic()
20
21
    # Methods will go here

1
async def connect_to_server(self, linked_account_token: str):
2
    """Connect to an MCP server
3
    Args:
4
        linked_account_token: The token for the associated Linked Account
5
    """
6
7
    server_params = StdioServerParameters(
8
        command="uvx",
9
        args=["merge-mcp"],
10
        env={
11
            "MERGE_API_KEY": os.getenv("MERGE_API_KEY"),
12
            "MERGE_ACCOUNT_TOKEN": linked_account_token
13
        }
14
    )
15
16
    stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
17
    self.stdio, self.write = stdio_transport
18
    self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
19
20
    await self.session.initialize()
21
22
    # List available tools
23
    response = await self.session.list_tools()
24
    tools = response.tools
25
    print("\nConnected to server with tools:", [tool.name for tool in tools])

1
async def process_query(self, query: str) -> str:
2
    """Process a query using Claude and available tools"""
3
    messages = [
4
        {
5
            "role": "user",
6
            "content": query
7
        }
8
    ]
9
10
    response = await self.session.list_tools()
11
    available_tools = [{
12
        "name": tool.name,
13
        "description": tool.description,
14
        "input_schema": tool.inputSchema
15
    } for tool in response.tools]
16
17
    # Initial Claude API call
18
    response = self.anthropic.messages.create(
19
        model="claude-3-5-sonnet-20241022",
20
        max_tokens=1000,
21
        messages=messages,
22
        tools=available_tools
23
    )
24
25
    # Process response and handle tool calls
26
    final_text = []
27
    assistant_message_content = []
28
    for content in response.content:
29
        if content.type == 'text':
30
            final_text.append(content.text)
31
            assistant_message_content.append(content)
32
33
        elif content.type == 'tool_use':
34
            tool_name = content.name
35
            tool_args = content.input
36
37
            # Get confirmation for tool call execution
38
            confirmation = input(f"Do you want to call tool '{tool_name}' with arguments {tool_args}? (y/n): ").strip().lower()
39
            if confirmation.startswith('y'):
40
                result = await self.session.call_tool(tool_name, tool_args)
41
                final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")
42
                assistant_message_content.append(content)
43
                messages.append({
44
                    "role": "assistant",
45
                    "content": assistant_message_content
46
                })
47
                messages.append({
48
                    "role": "user",
49
                    "content": [
50
                        {
51
                            "type": "tool_result",
52
                            "tool_use_id": content.id,
53
                            "content": result.content
54
                        }
55
                    ]
56
                })
57
58
                # Get next response from Claude
59
                response = self.anthropic.messages.create(
60
                    model="claude-3-5-sonnet-20241022",
61
                    max_tokens=1000,
62
                    messages=messages,
63
                    tools=available_tools
64
                )
65
                final_text.append(response.content[0].text)
66
67
            else:
68
                final_text.append(f"[Skipped calling tool {tool_name} with args {tool_args}]")
69
70
    return "\n".join(final_text)

1
async def chat_loop(self):
2
    """Run an interactive chat loop"""
3
    print("\nMCP Client Started!")
4
    print("Type your queries or 'quit' to exit.")
5
6
    def get_query():
7
        return input("\nQuery: ").strip()
8
9
    query = get_query()
10
11
    while query.lower() != 'quit':
12
        try:
13
            response = await self.process_query(query)
14
            print("\n" + response)
15
16
            query = get_query()
17
18
        except Exception as e:
19
            print(f"\nError: {str(e)}")

1
async def cleanup(self):
2
    """Clean up resources"""
3
    await self.exit_stack.aclose()

1
async def main():
2
    client = MCPClient()
3
    try:
4
        await client.connect_to_server("<your Linked Account token here>")
5
        await client.chat_loop()
6
    finally:
7
        await client.cleanup()
8
9
if __name__ == "__main__":
10
    import sys
11
    asyncio.run(main())

$
python client.py