mcp_client_example.py
1 #!/usr/bin/env python3 2 """ 3 MCP Client Example 4 5 Example of connecting to a PraisonAI MCP server as a client. 6 7 Prerequisites: 8 1. Start the MCP server first: 9 praisonai mcp serve --transport http-stream --port 8080 10 11 2. Then run this client: 12 python mcp_client_example.py 13 14 Environment Variables: 15 OPENAI_API_KEY - Required for chat completion tools 16 """ 17 18 import requests 19 20 21 def main(): 22 """Connect to MCP server and call tools.""" 23 base_url = "http://127.0.0.1:8080/mcp" 24 session_id = None 25 26 print("PraisonAI MCP Client Example") 27 print("=" * 40) 28 29 # 1. Initialize connection 30 print("\n1. Initializing connection...") 31 init_request = { 32 "jsonrpc": "2.0", 33 "id": 1, 34 "method": "initialize", 35 "params": { 36 "protocolVersion": "2025-11-25", 37 "capabilities": {}, 38 "clientInfo": { 39 "name": "example-client", 40 "version": "1.0.0", 41 }, 42 }, 43 } 44 45 response = requests.post( 46 base_url, 47 json=init_request, 48 headers={"Content-Type": "application/json"}, 49 ) 50 51 if response.status_code == 200: 52 result = response.json() 53 session_id = response.headers.get("MCP-Session-Id") 54 print(f" Connected! Session ID: {session_id}") 55 print(f" Server: {result.get('result', {}).get('serverInfo', {})}") 56 print(f" Protocol: {result.get('result', {}).get('protocolVersion')}") 57 else: 58 print(f" Error: {response.status_code}") 59 return 60 61 # 2. Send initialized notification 62 print("\n2. Sending initialized notification...") 63 init_notification = { 64 "jsonrpc": "2.0", 65 "method": "notifications/initialized", 66 } 67 requests.post( 68 base_url, 69 json=init_notification, 70 headers={ 71 "Content-Type": "application/json", 72 "MCP-Session-Id": session_id, 73 }, 74 ) 75 print(" Done") 76 77 # 3. List available tools 78 print("\n3. Listing available tools...") 79 list_request = { 80 "jsonrpc": "2.0", 81 "id": 2, 82 "method": "tools/list", 83 "params": {}, 84 } 85 86 response = requests.post( 87 base_url, 88 json=list_request, 89 headers={ 90 "Content-Type": "application/json", 91 "MCP-Session-Id": session_id, 92 }, 93 ) 94 95 if response.status_code == 200: 96 result = response.json() 97 tools = result.get("result", {}).get("tools", []) 98 print(f" Found {len(tools)} tools") 99 for tool in tools[:5]: # Show first 5 100 print(f" - {tool.get('name')}") 101 if len(tools) > 5: 102 print(f" ... and {len(tools) - 5} more") 103 104 # 4. List available resources 105 print("\n4. Listing available resources...") 106 list_resources = { 107 "jsonrpc": "2.0", 108 "id": 3, 109 "method": "resources/list", 110 "params": {}, 111 } 112 113 response = requests.post( 114 base_url, 115 json=list_resources, 116 headers={ 117 "Content-Type": "application/json", 118 "MCP-Session-Id": session_id, 119 }, 120 ) 121 122 if response.status_code == 200: 123 result = response.json() 124 resources = result.get("result", {}).get("resources", []) 125 print(f" Found {len(resources)} resources") 126 for res in resources: 127 print(f" - {res.get('uri')}") 128 129 # 5. List available prompts 130 print("\n5. Listing available prompts...") 131 list_prompts = { 132 "jsonrpc": "2.0", 133 "id": 4, 134 "method": "prompts/list", 135 "params": {}, 136 } 137 138 response = requests.post( 139 base_url, 140 json=list_prompts, 141 headers={ 142 "Content-Type": "application/json", 143 "MCP-Session-Id": session_id, 144 }, 145 ) 146 147 if response.status_code == 200: 148 result = response.json() 149 prompts = result.get("result", {}).get("prompts", []) 150 print(f" Found {len(prompts)} prompts") 151 for prompt in prompts: 152 print(f" - {prompt.get('name')}") 153 154 # 6. Read a resource 155 print("\n6. Reading MCP status resource...") 156 read_resource = { 157 "jsonrpc": "2.0", 158 "id": 5, 159 "method": "resources/read", 160 "params": { 161 "uri": "praisonai://mcp/status", 162 }, 163 } 164 165 response = requests.post( 166 base_url, 167 json=read_resource, 168 headers={ 169 "Content-Type": "application/json", 170 "MCP-Session-Id": session_id, 171 }, 172 ) 173 174 if response.status_code == 200: 175 result = response.json() 176 contents = result.get("result", {}).get("contents", []) 177 if contents: 178 print(f" Status: {contents[0].get('text', 'N/A')}") 179 180 # 7. Get a prompt 181 print("\n7. Getting deep-research prompt...") 182 get_prompt = { 183 "jsonrpc": "2.0", 184 "id": 6, 185 "method": "prompts/get", 186 "params": { 187 "name": "deep-research", 188 "arguments": { 189 "topic": "AI agents", 190 "depth": "medium", 191 }, 192 }, 193 } 194 195 response = requests.post( 196 base_url, 197 json=get_prompt, 198 headers={ 199 "Content-Type": "application/json", 200 "MCP-Session-Id": session_id, 201 }, 202 ) 203 204 if response.status_code == 200: 205 result = response.json() 206 messages = result.get("result", {}).get("messages", []) 207 if messages: 208 content = messages[0].get("content", {}) 209 text = content.get("text", "") if isinstance(content, dict) else str(content) 210 print(f" Prompt preview: {text[:100]}...") 211 212 # 8. Ping 213 print("\n8. Pinging server...") 214 ping_request = { 215 "jsonrpc": "2.0", 216 "id": 7, 217 "method": "ping", 218 "params": {}, 219 } 220 221 response = requests.post( 222 base_url, 223 json=ping_request, 224 headers={ 225 "Content-Type": "application/json", 226 "MCP-Session-Id": session_id, 227 }, 228 ) 229 230 if response.status_code == 200: 231 print(" Pong!") 232 233 print("\n" + "=" * 40) 234 print("MCP client example completed successfully!") 235 236 237 if __name__ == "__main__": 238 main()