diff --git a/mcp-client-python/client.py b/mcp-client-python/client.py index c53b39b1..e35b92dd 100644 --- a/mcp-client-python/client.py +++ b/mcp-client-python/client.py @@ -7,9 +7,11 @@ from anthropic import Anthropic from dotenv import load_dotenv +from pathlib import Path load_dotenv() # load environment variables from .env + class MCPClient: def __init__(self): # Initialize session and client objects @@ -19,28 +21,37 @@ def __init__(self): 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') + 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 + + if is_python: + path = Path(server_script_path).resolve() + server_params = StdioServerParameters( + command="uv", + args=["--directory", path.parent, "run", path.name], + env=None, + ) + else: + server_params = StdioServerParameters( + command="node", args=[server_script_path], env=None + ) + + stdio_transport = await self.exit_stack.enter_async_context( + stdio_client(server_params) ) - - 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)) - + 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 @@ -48,52 +59,44 @@ async def connect_to_server(self, server_script_path: str): async def process_query(self, query: str) -> str: """Process a query using Claude and available tools""" - messages = [ + messages = [{"role": "user", "content": query}] + + response = await self.session.list_tools() + available_tools = [ { - "role": "user", - "content": query + "name": tool.name, + "description": tool.description, + "input_schema": tool.inputSchema, } + for tool in response.tools ] - 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-sonnet-4-0", max_tokens=1000, messages=messages, - tools=available_tools + tools=available_tools, ) # Process response and handle tool calls final_text = [] for content in response.content: - if content.type == 'text': + if content.type == "text": final_text.append(content.text) - elif content.type == 'tool_use': + 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) 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 - }) + 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( @@ -110,29 +113,30 @@ 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': + + 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 ") sys.exit(1) - + client = MCPClient() try: await client.connect_to_server(sys.argv[1]) @@ -140,6 +144,8 @@ async def main(): finally: await client.cleanup() + if __name__ == "__main__": import sys + asyncio.run(main())