"""Example HTTP client for the Tool Search MCP Server.
This demonstrates how to use the REST API for tool management.
Usage:
# Set your API key
export API_KEY="your-api-key"
export SERVER_URL="https://your-app.herokuapp.com"
# Run examples
python example_clients/http_client.py list
python example_clients/http_client.py register
python example_clients/http_client.py search_bm25 "weather"
python example_clients/http_client.py search_semantic "check temperature"
"""
import os
import sys
import httpx
import json
# Configuration
SERVER_URL = os.getenv("SERVER_URL", "http://localhost:8000")
API_KEY = os.getenv("API_KEY", "")
def get_headers():
"""Get request headers with authentication."""
headers = {"Content-Type": "application/json"}
if API_KEY:
headers["Authorization"] = f"Bearer {API_KEY}"
return headers
def list_tools():
"""List all tools in the catalog."""
response = httpx.get(f"{SERVER_URL}/tools", headers=get_headers())
response.raise_for_status()
tools = response.json()
print(f"Found {len(tools)} tools:")
for tool in tools:
print(f" - {tool['name']}: {tool['description'][:60]}...")
def register_sample_tool():
"""Register a sample tool."""
tool = {
"name": "sample_tool",
"description": "A sample tool registered via the API",
"input_schema": {
"type": "object",
"properties": {
"message": {
"type": "string",
"description": "A message to process"
}
},
"required": ["message"]
},
"tags": ["sample", "demo"]
}
response = httpx.post(
f"{SERVER_URL}/tools",
headers=get_headers(),
json=tool
)
response.raise_for_status()
result = response.json()
print("Registered tool:")
print(json.dumps(result, indent=2))
def search_bm25(query: str):
"""Search tools using BM25."""
response = httpx.post(
f"{SERVER_URL}/search/bm25",
headers=get_headers(),
json={"query": query, "top_k": 5}
)
response.raise_for_status()
results = response.json()
print(f"BM25 search results for '{query}':")
for tool in results["results"]:
print(f" - {tool['name']}: {tool['description'][:60]}...")
def search_semantic(query: str):
"""Search tools using semantic similarity."""
response = httpx.post(
f"{SERVER_URL}/search/semantic",
headers=get_headers(),
json={"query": query, "top_k": 5}
)
response.raise_for_status()
results = response.json()
print(f"Semantic search results for '{query}':")
for tool in results["results"]:
score = tool.get("score", "N/A")
print(f" - {tool['name']} (score: {score:.3f}): {tool['description'][:50]}...")
def search_regex(pattern: str):
"""Search tools using regex pattern."""
response = httpx.post(
f"{SERVER_URL}/search/regex",
headers=get_headers(),
json={"pattern": pattern, "top_k": 5}
)
response.raise_for_status()
results = response.json()
print(f"Regex search results for '{pattern}':")
for tool in results["results"]:
print(f" - {tool['name']}: {tool['description'][:60]}...")
def health_check():
"""Check server health."""
response = httpx.get(f"{SERVER_URL}/health")
response.raise_for_status()
result = response.json()
print(f"Server status: {result['status']}")
print(f"Tool count: {result['tool_count']}")
def main():
"""Run the HTTP client."""
if len(sys.argv) < 2:
print("Usage: python http_client.py <command> [args]")
print("\nCommands:")
print(" health - Check server health")
print(" list - List all tools")
print(" register - Register a sample tool")
print(" search_bm25 <query> - Search using BM25")
print(" search_regex <pattern> - Search using regex")
print(" search_semantic <query> - Search using semantic similarity")
print(f"\nServer URL: {SERVER_URL}")
print(f"API Key configured: {'Yes' if API_KEY else 'No'}")
sys.exit(1)
command = sys.argv[1]
try:
if command == "health":
health_check()
elif command == "list":
list_tools()
elif command == "register":
register_sample_tool()
elif command == "search_bm25":
if len(sys.argv) < 3:
print("Error: search_bm25 requires a query argument")
sys.exit(1)
search_bm25(sys.argv[2])
elif command == "search_regex":
if len(sys.argv) < 3:
print("Error: search_regex requires a pattern argument")
sys.exit(1)
search_regex(sys.argv[2])
elif command == "search_semantic":
if len(sys.argv) < 3:
print("Error: search_semantic requires a query argument")
sys.exit(1)
search_semantic(sys.argv[2])
else:
print(f"Unknown command: {command}")
sys.exit(1)
except httpx.HTTPStatusError as e:
print(f"HTTP Error: {e.response.status_code}")
print(e.response.text)
sys.exit(1)
except httpx.ConnectError:
print(f"Error: Could not connect to server at {SERVER_URL}")
sys.exit(1)
if __name__ == "__main__":
main()
