1  #!/usr/bin/env python3
  2  """
  3  SSE Test Client for Ultimate MCP Server
  4  Tests server functionality over SSE (Server-Sent Events) transport
  5  """
  6  
  7  import asyncio
  8  import json
  9  
 10  from fastmcp import Client
 11  
 12  
 13  async def test_sse_server():
 14      """Test Ultimate MCP Server over SSE transport."""
 15      # SSE endpoint - note the /sse path for SSE transport
 16      server_url = "http://127.0.0.1:8013/sse"
 17  
 18      print("🔥 Ultimate MCP Server SSE Test Client")
 19      print("=" * 50)
 20      print(f"🔗 Connecting to Ultimate MCP Server SSE endpoint at {server_url}")
 21  
 22      try:
 23          async with Client(server_url) as client:
 24              print("✅ Successfully connected to SSE server")
 25  
 26              # Test 1: List available tools
 27              print("\n📋 Testing tool discovery...")
 28              tools = await client.list_tools()
 29              print(f"Found {len(tools)} tools via SSE transport:")
 30              for i, tool in enumerate(tools[:10]):  # Show first 10
 31                  print(f"  {i + 1:2d}. {tool.name}")
 32              if len(tools) > 10:
 33                  print(f"  ... and {len(tools) - 10} more tools")
 34  
 35              # Test 2: List available resources
 36              print("\n📚 Testing resource discovery...")
 37              resources = await client.list_resources()
 38              print(f"Found {len(resources)} resources:")
 39              for resource in resources:
 40                  print(f"  - {resource.uri}")
 41  
 42              # Test 3: Echo tool test
 43              print("\n🔊 Testing echo tool over SSE...")
 44              echo_result = await client.call_tool("echo", {"message": "Hello from SSE client!"})
 45              if echo_result:
 46                  echo_data = json.loads(echo_result[0].text)
 47                  print(f"✅ Echo response: {json.dumps(echo_data, indent=2)}")
 48  
 49              # Test 4: Provider status test
 50              print("\n🔌 Testing provider status over SSE...")
 51              try:
 52                  provider_result = await client.call_tool("get_provider_status", {})
 53                  if provider_result:
 54                      provider_data = json.loads(provider_result[0].text)
 55                      providers = provider_data.get("providers", {})
 56                      print(f"✅ Found {len(providers)} providers via SSE:")
 57                      for name, status in providers.items():
 58                          available = "✅" if status.get("available") else "❌"
 59                          model_count = len(status.get("models", []))
 60                          print(f"  {available} {name}: {model_count} models")
 61              except Exception as e:
 62                  print(f"❌ Provider status failed: {e}")
 63  
 64              # Test 5: Resource reading test
 65              print("\n📖 Testing resource reading over SSE...")
 66              if resources:
 67                  try:
 68                      resource_uri = resources[0].uri
 69                      resource_content = await client.read_resource(resource_uri)
 70                      if resource_content:
 71                          content = resource_content[0].text
 72                          preview = content[:200] + "..." if len(content) > 200 else content
 73                          print(f"✅ Resource {resource_uri} content preview:")
 74                          print(f"  {preview}")
 75                  except Exception as e:
 76                      print(f"❌ Resource reading failed: {e}")
 77  
 78              # Test 6: Simple completion test (if providers available)
 79              print("\n🤖 Testing completion over SSE...")
 80              try:
 81                  completion_result = await client.call_tool(
 82                      "generate_completion",
 83                      {
 84                          "prompt": "Say hello in exactly 3 words",
 85                          "provider": "ollama",
 86                          "model": "mix_77/gemma3-qat-tools:27b",
 87                          "max_tokens": 10,
 88                      },
 89                  )
 90                  if completion_result:
 91                      result_data = json.loads(completion_result[0].text)
 92                      print("✅ Completion via SSE:")
 93                      print(f"  Text: '{result_data.get('text', 'No text')}'")
 94                      print(f"  Model: {result_data.get('model', 'Unknown')}")
 95                      print(f"  Success: {result_data.get('success', False)}")
 96                      print(f"  Processing time: {result_data.get('processing_time', 0):.2f}s")
 97              except Exception as e:
 98                  print(f"⚠️ Completion test failed (expected if no providers): {e}")
 99  
100              # Test 7: Filesystem tool test
101              print("\n📁 Testing filesystem tools over SSE...")
102              try:
103                  dirs_result = await client.call_tool("list_allowed_directories", {})
104                  if dirs_result:
105                      dirs_data = json.loads(dirs_result[0].text)
106                      print(
107                          f"✅ Allowed directories via SSE: {dirs_data.get('count', 0)} directories"
108                      )
109              except Exception as e:
110                  print(f"❌ Filesystem test failed: {e}")
111  
112              # Test 8: Text processing tool test
113              print("\n📝 Testing text processing over SSE...")
114              try:
115                  ripgrep_result = await client.call_tool(
116                      "run_ripgrep", {"args_str": "'async' . -t py --max-count 5", "input_dir": "."}
117                  )
118                  if ripgrep_result:
119                      ripgrep_data = json.loads(ripgrep_result[0].text)
120                      if ripgrep_data.get("success"):
121                          lines = ripgrep_data.get("output", "").split("\n")
122                          line_count = len([l for l in lines if l.strip()])  # noqa: E741
123                          print(f"✅ Ripgrep via SSE found {line_count} matching lines")
124                      else:
125                          print("⚠️ Ripgrep completed but found no matches")
126              except Exception as e:
127                  print(f"❌ Text processing test failed: {e}")
128  
129              print("\n🎉 SSE transport functionality test completed!")
130              return True
131  
132      except Exception as e:
133          print(f"❌ SSE connection failed: {e}")
134          print("\nTroubleshooting:")
135          print("1. Make sure the server is running in SSE mode:")
136          print("   umcp run -t sse")
137          print("2. Verify the server is accessible at http://127.0.0.1:8013")
138          print("3. Check that the SSE endpoint is available at /sse")
139          return False
140  
141  
142  async def test_sse_interactive():
143      """Interactive SSE testing mode."""
144      server_url = "http://127.0.0.1:8013/sse"
145  
146      print("\n🎮 Entering SSE interactive mode...")
147      print("Type 'list' to see available tools, 'quit' to exit")
148  
149      try:
150          async with Client(server_url) as client:
151              tools = await client.list_tools()
152              resources = await client.list_resources()
153  
154              while True:
155                  try:
156                      command = input("\nSSE> ").strip()
157  
158                      if command.lower() in ["quit", "exit", "q"]:
159                          print("👋 Goodbye!")
160                          break
161                      elif command.lower() == "list":
162                          print("Available tools:")
163                          for i, tool in enumerate(tools[:20]):
164                              print(f"  {i + 1:2d}. {tool.name}")
165                          if len(tools) > 20:
166                              print(f"  ... and {len(tools) - 20} more")
167                      elif command.lower() == "resources":
168                          print("Available resources:")
169                          for resource in resources:
170                              print(f"  - {resource.uri}")
171                      elif command.startswith("tool "):
172                          # Call tool: tool <tool_name> <json_params>
173                          parts = command[5:].split(" ", 1)
174                          tool_name = parts[0]
175                          params = json.loads(parts[1]) if len(parts) > 1 else {}
176  
177                          try:
178                              result = await client.call_tool(tool_name, params)
179                              if result:
180                                  print(f"✅ Tool result: {result[0].text}")
181                              else:
182                                  print("❌ No result returned")
183                          except Exception as e:
184                              print(f"❌ Tool call failed: {e}")
185                      elif command.startswith("read "):
186                          # Read resource: read <resource_uri>
187                          resource_uri = command[5:].strip()
188                          try:
189                              result = await client.read_resource(resource_uri)
190                              if result:
191                                  content = result[0].text
192                                  preview = content[:500] + "..." if len(content) > 500 else content
193                                  print(f"✅ Resource content: {preview}")
194                              else:
195                                  print("❌ No content returned")
196                          except Exception as e:
197                              print(f"❌ Resource read failed: {e}")
198                      else:
199                          print("Commands:")
200                          print("  list          - List available tools")
201                          print("  resources     - List available resources")
202                          print("  tool <name> <params>  - Call a tool with JSON params")
203                          print("  read <uri>    - Read a resource")
204                          print("  quit          - Exit interactive mode")
205  
206                  except KeyboardInterrupt:
207                      print("\n👋 Goodbye!")
208                      break
209                  except Exception as e:
210                      print(f"❌ Command error: {e}")
211  
212      except Exception as e:
213          print(f"❌ SSE interactive mode failed: {e}")
214  
215  
216  async def main():
217      """Main test function."""
218      # Run basic functionality test
219      success = await test_sse_server()
220  
221      if success:
222          # Ask if user wants interactive mode
223          try:
224              response = (
225                  input("\nWould you like to enter SSE interactive mode? (y/n): ").strip().lower()
226              )
227              if response in ["y", "yes"]:
228                  await test_sse_interactive()
229          except KeyboardInterrupt:
230              print("\n👋 Goodbye!")
231  
232  
233  if __name__ == "__main__":
234      asyncio.run(main())