client_example_python.py

"""
Complete Python MCP Client Example

This client demonstrates how to:
1. Connect to an MCP server using stdio transport
2. List available tools and resources
3. Call calculator tools
4. Handle responses from the server
"""

import asyncio
import json
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client


class MCPCalculatorClient:
    def __init__(self):
        # Create server parameters for stdio connection
        self.server_params = StdioServerParameters(
            command="python",  # Executable
            args=["../01-first-server/solution/python/server.py"],  # Server script
            env=None,  # Optional environment variables
        )

    async def run(self):
        """Main client execution function"""
        print("🚀 Starting MCP Python Client...")

        try:
            async with stdio_client(self.server_params) as (read, write):
                async with ClientSession(read, write) as session:
                    print("📡 Connecting to MCP server...")
                    
                    # Initialize the connection
                    await session.initialize()
                    print("✅ Connected to MCP server successfully!")

                    # List available tools
                    await self.list_tools(session)
                    
                    # Test calculator operations
                    await self.test_calculator_operations(session)
                    
                    # List and test resources
                    await self.list_and_test_resources(session)
                    
                    print("\n✨ Client operations completed successfully!")

        except Exception as e:
            print(f"❌ Error running MCP client: {e}")
            raise

    async def list_tools(self, session: ClientSession):
        """List all available tools on the server"""
        print("\n📋 Listing available tools:")
        try:
            tools = await session.list_tools()
            for tool in tools.tools:
                print(f"  - {tool.name}: {tool.description}")
        except Exception as e:
            print(f"  Error listing tools: {e}")

    async def test_calculator_operations(self, session: ClientSession):
        """Test various calculator operations"""
        print("\n🧮 Testing Calculator Operations:")

        operations = [
            ("add", {"a": 5, "b": 3}, "Add 5 + 3"),
            ("subtract", {"a": 10, "b": 4}, "Subtract 10 - 4"),
            ("multiply", {"a": 6, "b": 7}, "Multiply 6 × 7"),
            ("divide", {"a": 20, "b": 4}, "Divide 20 ÷ 4"),
            ("help", {}, "Help Information"),
        ]

        for tool_name, arguments, description in operations:
            try:
                result = await session.call_tool(tool_name, arguments=arguments)
                result_text = self.extract_text_result(result)
                
                if tool_name == "help":
                    print(f"\n📖 {description}:")
                    print(result_text)
                else:
                    print(f"{description} = {result_text}")
                    
            except Exception as e:
                print(f"  Error calling {tool_name}: {e}")

    async def list_and_test_resources(self, session: ClientSession):
        """List and test reading resources"""
        print("\n📄 Listing available resources:")
        try:
            resources = await session.list_resources()
            for resource in resources.resources:
                print(f"  - {resource.name}: {resource.description}")
                print(f"    URI: {resource.uri}")

            # Test reading a resource if available
            if resources.resources:
                first_resource = resources.resources[0]
                print(f"\n📖 Reading resource: {first_resource.name}")
                try:
                    content = await session.read_resource(first_resource.uri)
                    print(f"Resource content: {content}")
                except Exception as e:
                    print(f"  Error reading resource: {e}")
            else:
                print("  No resources available")
                
        except Exception as e:
            print(f"  Error listing resources: {e}")

    def extract_text_result(self, result) -> str:
        """
        Extract text content from a tool result object.

        This method attempts to extract the text content from the `content` attribute
        of the result object. If no text content is found, it falls back to converting
        the result to a string. If an error occurs during extraction, it returns "No result".

        Args:
            result: The result object returned by a tool, which may contain a `content` attribute
                    with text or other types of data.

        Returns:
            A string representing the extracted text content, or a fallback string if no text is found.
        """
        try:
            if hasattr(result, 'content') and result.content:
                for content_item in result.content:
                    if hasattr(content_item, 'text') and content_item.text:
                        return content_item.text
                    elif hasattr(content_item, 'type') and content_item.type == "text":
                        return getattr(content_item, 'text', str(content_item))
            
            # Fallback: try to convert to string
            return str(result)
        except Exception:
            return "No result"


async def main():
    """Entry point for the client"""
    client = MCPCalculatorClient()
    await client.run()


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