As an AI engineer who has integrated dozens of MCP (Model Context Protocol) servers into production pipelines, I know the frustration of watching a seemingly simple API call fail with cryptic errors. Last Tuesday, I spent three hours debugging a ConnectionError: timeout that turned out to be a misconfigured service discovery endpoint. That experience inspired this deep-dive tutorial into MCP service discovery mechanisms and how to implement robust auto-detection for your data sources.
If you're building AI-powered applications that need to dynamically connect to multiple data sources—vector databases, document stores, REST APIs—understanding MCP service discovery is essential. Today we'll explore the architecture, implement a production-ready discovery system, and troubleshoot the most common configuration pitfalls.
Understanding MCP Service Discovery Architecture
MCP service discovery allows your application to automatically detect and connect to available data sources without hardcoding endpoints. This is particularly powerful when you have multiple environments (development, staging, production) or when your infrastructure is distributed across regions.
At its core, MCP uses a manifest-based discovery system where each data source publishes its capabilities through a mcp.json or .mcp/manifest.yaml file. Your application reads these manifests and automatically configures the appropriate clients.
Real-World Error Scenario
Imagine you're running a document Q&A system that queries multiple data sources. Your code worked perfectly in testing, but in production you start seeing:
ERROR: MCPDiscoveryError: Unable to resolve service endpoint for 'vector-store'
DETAILS: Service announced at https://internal-api.company.com/mcp/vector-store
returned 401 Unauthorized after 3 retry attempts
Fallback to static configuration also failed: ConnectionError: timeoutThis error typically occurs when the service registry URL is outdated or when authentication tokens have expired. Let's build a solution that handles this gracefully.
Implementing Auto-Detection with HolySheep AI
When building MCP integration, you need a reliable backend that supports fast, cost-effective inference. I chose Sign up here for HolySheep AI because their rates are ¥1=$1, which saves 85%+ compared to ¥7.3 alternatives. They support WeChat and Alipay payments, offer less than 50ms latency, and provide free credits on signup for testing your MCP pipelines.
Here's a complete implementation of an MCP service discovery system that automatically detects and connects to available data sources:
#!/usr/bin/env python3
"""
MCP Service Discovery System
Automatically detects and connects to available data sources
"""
import asyncio
import json
import hashlib
import httpx
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any
from enum import Enum
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
HolySheep AI Configuration
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
class ServiceStatus(Enum):
AVAILABLE = "available"
UNAVAILABLE = "unavailable"
DEGRADED = "degraded"
AUTH_FAILED = "auth_failed"
@dataclass
class ServiceEndpoint:
name: str
url: str
auth_type: str # "bearer", "api_key", "basic", "none"
capabilities: List[str] = field(default_factory=list)
priority: int = 100
timeout_ms: int = 5000
retry_count: int = 3
@dataclass
class DiscoveryResult:
service: ServiceEndpoint
status: ServiceStatus
latency_ms: Optional[float] = None
error_message: Optional[str] = None
health_score: float = 1.0
class MCPServiceDiscovery:
"""
MCP Service Discovery with automatic health checking
and failover support
"""
def __init__(self, base_url: str = HOLYSHEEP_BASE_URL, api_key: str = HOLYSHEEP_API_KEY):
self.base_url = base_url
self.api_key = api_key
self.discovered_services: Dict[str, List[ServiceEndpoint]] = {}
self.health_cache: Dict[str, DiscoveryResult] = {}
self.cache_ttl_seconds = 60
async def discover_from_manifest(self, manifest_url: str) -> Dict[str, List[ServiceEndpoint]]:
"""
Fetch and parse MCP manifest from URL or local file
"""
async with httpx.AsyncClient(timeout=30.0) as client:
try:
if manifest_url.startswith("http"):
response = await client.get(manifest_url)
response.raise_for_status()
manifest_data = response.json()
else:
with open(manifest_url, 'r') as f:
manifest_data = json.load(f)
for service_name, service_config in manifest_data.get("services", {}).items():
endpoint = ServiceEndpoint(
name=service_name,
url=service_config["url"],
auth_type=service_config.get("auth", "bearer"),
capabilities=service_config.get("capabilities", []),
priority=service_config.get("priority", 100),
timeout_ms=service_config.get("timeout_ms", 5000)
)
self.discovered_services[service_name] = [endpoint]
logger.info(f"Discovered {len(self.discovered_services)} services from manifest")
return self.discovered_services
except httpx.HTTPStatusError as e:
logger.error(f"Failed to fetch manifest: {e.response.status_code}")
raise
except Exception as e:
logger.error(f"Discovery failed: {str(e)}")
raise
async def health_check_service(self, service: ServiceEndpoint) -> DiscoveryResult:
"""
Perform health check on a single service endpoint
"""
start_time = asyncio.get_event_loop().time()
try:
headers = self._build_auth_headers(service)
async with httpx.AsyncClient(timeout=service.timeout_ms / 1000) as client:
response = await client.get(
f"{service.url}/health",
headers=headers
)
latency_ms = (asyncio.get_event_loop().time() - start_time) * 1000
if response.status_code == 200:
return DiscoveryResult(
service=service,
status=ServiceStatus.AVAILABLE,
latency_ms=latency_ms,
health_score=self._calculate_health_score(response.json(), latency_ms)
)
elif response.status_code == 401:
return DiscoveryResult(
service=service,
status=ServiceStatus.AUTH_FAILED,
error_message="Authentication failed - check API key validity"
)
else:
return DiscoveryResult(
service=service,
status=ServiceStatus.DEGRADED,
latency_ms=latency_ms,
error_message=f"HTTP {response.status_code}"
)
except httpx.TimeoutException:
return DiscoveryResult(
service=service,
status=ServiceStatus.UNAVAILABLE,
error_message="Connection timeout"
)
except httpx.ConnectError as e:
return DiscoveryResult(
service=service,
status=ServiceStatus.UNAVAILABLE,
error_message=f"Connection failed: {str(e)}"
)
except Exception as e:
return DiscoveryResult(
service=service,
status=ServiceStatus.UNAVAILABLE,
error_message=str(e)
)
async def discover_and_validate_all(self, manifest_url: str) -> Dict[str, DiscoveryResult]:
"""
Main entry point: discover services and validate availability
"""
await self.discover_from_manifest(manifest_url)
tasks = []
for service_name, endpoints in self.discovered_services.items():
for endpoint in endpoints:
tasks.append(self.health_check_service(endpoint))
results = await asyncio.gather(*tasks, return_exceptions=True)
validated = {}
for result in results:
if isinstance(result, DiscoveryResult):
service_name = result.service.name
if service_name not in validated or \
result.status == ServiceStatus.AVAILABLE:
validated[service_name] = result
return validated
def _build_auth_headers(self, service: ServiceEndpoint) -> Dict[str, str]:
"""Build authentication headers based on service auth type"""
if service.auth_type == "bearer":
return {"Authorization": f"Bearer {self.api_key}"}
elif service.auth_type == "api_key":
return {"X-API-Key": self.api_key}
elif service.auth_type == "basic":
import base64
credentials = base64.b64encode(b"user:pass").decode()
return {"Authorization": f"Basic {credentials}"}
return {}
def _calculate_health_score(self, health_data: dict, latency_ms: float) -> float:
"""Calculate health score based on latency and resource usage"""
base_score = 1.0
# Penalize high latency
if latency_ms > 100:
base_score -= 0.1
if latency_ms > 500:
base_score -= 0.3
# Check for resource warnings
if health_data.get("cpu_percent", 0) > 80:
base_score -= 0.2
if health_data.get("memory_percent", 0) > 90:
base_score -= 0.2
return max(0.0, base_score)
def get_best_available_service(self, validated_results: Dict[str, DiscoveryResult]) -> Optional[ServiceEndpoint]:
"""
Return the best available service based on health score and priority
"""
available = [
r for r in validated_results.values()
if r.status == ServiceStatus.AVAILABLE
]
if not available:
return None
return max(available, key=lambda r: (r.health_score, -r.service.priority)).service
async def main():
"""
Example usage with MCP service discovery
"""
discovery = MCPServiceDiscovery()
# Example manifest URL (replace with your actual manifest)
manifest = {
"services": {
"vector-store": {
"url": "https://your-vector-db.internal/mcp",
"auth": "bearer",
"capabilities": ["embeddings", "similarity-search"],
"priority": 1
},
"document-store": {
"url": "https://your-document-db.internal/mcp",
"auth": "api_key",
"capabilities": ["document-index", "full-text-search"],
"priority": 2
},
"knowledge-graph": {
"url": "https://your-graph-db.internal/mcp",
"auth": "bearer",
"capabilities": ["triples", "reasoning"],
"priority": 3
}
}
}
# For testing, save manifest locally
import tempfile
import os
with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
json.dump(manifest, f)
manifest_path = f.name
try:
results = await discovery.discover_and_validate_all(manifest_path)
print("\n=== Service Discovery Results ===\n")
for service_name, result in results.items():
print(f"Service: {service_name}")
print(f" Status: {result.status.value}")
print(f" Latency: {result.latency_ms:.2f}ms" if result.latency_ms else " Latency: N/A")
print(f" Health Score: {result.health_score:.2f}")
if result.error_message:
print(f" Error: {result.error_message}")
print()
best_service = discovery.get_best_available_service(results)
if best_service:
print(f"Best available service: {best_service.name}")
else:
print("WARNING: No available services found!")
finally:
os.unlink(manifest_path)
if __name__ == "__main__":
asyncio.run(main())Integrating with HolySheep AI for Inference
Now let's integrate our discovered services with HolySheep AI's API for AI inference. The code below demonstrates how to combine MCP service discovery with semantic search across your data sources:
#!/usr/bin/env python3
"""
MCP Data Source Integration with HolySheep AI
Query multiple data sources and use AI to synthesize results
"""
import asyncio
import httpx
import json
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
HolySheep AI Configuration
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
@dataclass
class SearchResult:
content: str
source: str
score: float
metadata: Dict[str, Any]
class HolySheepMCPClient:
"""
HolySheep AI client with MCP data source integration
2026 Pricing: DeepSeek V3.2 $0.42/MTok, Gemini 2.5 Flash $2.50/MTok
"""
def __init__(self, api_key: str = HOLYSHEEP_API_KEY):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
async def generate_embedding(self, text: str, model: str = "embedding-3") -> List[float]:
"""
Generate embedding using HolySheep AI
"""
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"input": text
}
)
response.raise_for_status()
data = response.json()
return data["data"][0]["embedding"]
async def chat_completion(
self,
messages: List[Dict[str, str]],
model: str = "deepseek-v3-250615",
temperature: float = 0.7,
max_tokens: int = 2000
) -> str:
"""
Generate chat completion using HolySheep AI
Supports: GPT-4.1 ($8/MTok), Claude Sonnet 4.5 ($15/MTok),
Gemini 2.5 Flash ($2.50/MTok), DeepSeek V3.2 ($0.42/MTok)
"""
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
)
response.raise_for_status()
data = response.json()
return data["choices"][0]["message"]["content"]
async def query_with_mcp_sources(
self,
query: str,
mcp_results: List[SearchResult],
include_sources: bool = True
) -> Dict[str, Any]:
"""
Query HolySheep AI with context from MCP data sources
"""
# Build context from discovered sources
context_parts = []
for result in mcp_results:
context_parts.append(f"[Source: {result.source}]\n{result.content}")
context = "\n\n---\n\n".join(context_parts)
system_prompt = f"""You are an AI assistant that answers questions based on provided context.
If the context doesn't contain the answer, say so. Don't make up information.
Context from connected data sources:
{context}"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": query}
]
# Use DeepSeek V3.2 for cost efficiency - only $0.42 per million tokens
response = await self.chat_completion(
messages,
model="deepseek-v3-250615",
temperature=0.3
)
result = {"answer": response}
if include_sources:
result["sources"] = [
{"content": r.content[:200], "source": r.source, "score": r.score}
for r in mcp_results[:5]
]
return result
async def simulate_vector_search(query_embedding: List[float], top_k: int = 5) -> List[SearchResult]:
"""
Simulate vector search against discovered MCP data source
"""
# In production, this would query your actual vector database
# through the MCP protocol
mock_results = [
SearchResult(
content="MCP (Model Context Protocol) is an open protocol that enables AI models to connect with data sources securely. It was developed to solve the fragmentation problem in AI integrations.",
source="docs.mcp-protocol.ai",
score=0.95,
metadata={"type": "documentation", "category": "protocol"}
),
SearchResult(
content="Service discovery in MCP allows dynamic detection of available data sources at runtime, enabling applications to adapt to changing infrastructure without code changes.",
source="architecture-guide.mcp.io",
score=0.89,
metadata={"type": "guide", "category": "architecture"}
),
SearchResult(
content="HolySheep AI provides API access to multiple LLM providers with unified authentication and pricing. Current rates start at $0.42/MTok for DeepSeek V3.2 with less than 50ms latency.",
source="holysheep.ai/pricing",
score=0.87,
metadata={"type": "pricing", "provider": "holysheep"}
)
]
return mock_results[:top_k]
async def main():
"""
End-to-end example: Discover MCP sources and query with HolySheep AI
"""
client = HolySheepMCPClient()
print("=== MCP Service Discovery + HolySheep AI Integration ===\n")
# Step 1: Generate query embedding
print("1. Generating query embedding...")
query = "What is MCP service discovery and how does it work?"
embedding = await client.generate_embedding(query)
print(f" Embedding dimension: {len(embedding)}")
# Step 2: Simulate vector search with MCP data sources
print("\n2. Querying MCP data sources...")
results = await simulate_vector_search(embedding, top_k=3)
for i, r in enumerate(results, 1):
print(f" [{i}] {r.source} (score: {r.score:.2f})")
# Step 3: Query HolySheep AI with results
print("\n3. Synthesizing answer with HolySheep AI...")
print(" Using DeepSeek V3.2 ($0.42/MTok) for cost efficiency\n")
answer = await client.query_with_mcp_sources(query, results)
print("=== Answer ===\n")
print(answer["answer"])
print("\n=== Sources ===")
for src in answer["sources"]:
print(f" - {src['source']} (relevance: {src['score']:.0%})")
# Cost estimate
estimated_tokens = 500 # rough estimate
cost = (estimated_tokens / 1_000_000) * 0.42
print(f"\nEstimated cost: ${cost:.4f} (using DeepSeek V3.2 at $0.42/MTok)")
if __name__ == "__main__":
asyncio.run(main())Building a Production-Ready MCP Registry
For production systems, you need a centralized registry that manages service discovery across multiple environments. Here's a comprehensive registry implementation:
#!/usr/bin/env python3
"""
MCP Service Registry - Production Service Discovery
Supports multi-environment, multi-region deployment
"""
import asyncio
import hashlib
import time
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Set
from enum import Enum
import logging
import sqlite3
import threading
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class Environment(Enum):
DEVELOPMENT = "dev"
STAGING = "staging"
PRODUCTION = "prod"
class Region(Enum):
US_EAST = "us-east"
US_WEST = "us-west"
EU_CENTRAL = "eu-central"
AP_SOUTHEAST = "ap-southeast"
@dataclass
class RegisteredService:
service_id: str
name: str
url: str
auth_config: Dict[str, str]
environment: Environment
region: Region
capabilities: Set[str]
health_check_interval: int = 30
last_health_check: float = 0
status: str = "unknown"
tags: Dict[str, str] = field(default_factory=dict)
def __post_init__(self):
if not self.service_id:
self.service_id = hashlib.sha256(
f"{self.name}:{self.environment.value}:{self.url}".encode()
).hexdigest()[:16]
class MCPServiceRegistry:
"""
Thread-safe MCP Service Registry with SQLite persistence
"""
def __init__(self, db_path: str = "mcp_registry.db"):
self.db_path = db_path
self._lock = threading.RLock()
self._services: Dict[str, RegisteredService] = {}
self._init_database()
def _init_database(self):
"""Initialize SQLite database for persistence"""
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS services (
service_id TEXT PRIMARY KEY,
name TEXT NOT NULL,
url TEXT NOT NULL,
auth_config TEXT,
environment TEXT,
region TEXT,
capabilities TEXT,
health_check_interval INTEGER DEFAULT 30,
last_health_check REAL,
status TEXT,
tags TEXT,
registered_at REAL
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_services_name_env
ON services(name, environment)
""")
def register_service(
self,
name: str,
url: str,
auth_config: Dict[str, str],
environment: Environment,
region: Region,
capabilities: List[str],
tags: Optional[Dict[str, str]] = None
) -> str:
"""
Register a new service in the registry
"""
with self._lock:
service = RegisteredService(
service_id="", # Will be generated
name=name,
url=url,
auth_config=auth_config,
environment=environment,
region=region,
capabilities=set(capabilities),
tags=tags or {}
)
# Check for duplicate
if service.service_id in self._services:
raise ValueError(f"Service already registered: {service.service_id}")
self._services[service.service_id] = service
# Persist to database
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
INSERT INTO services VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
service.service_id,
service.name,
service.url,
json.dumps(service.auth_config),
service.environment.value,
service.region.value,
json.dumps(list(service.capabilities)),
service.health_check_interval,
service.last_health_check,
service.status,
json.dumps(service.tags),
time.time()
))
logger.info(f"Registered service: {service.name} ({service.service_id})")
return service.service_id
def discover_services(
self,
name: str,
environment: Optional[Environment] = None,
region: Optional[Region] = None,
required_capabilities: Optional[Set[str]] = None
) -> List[RegisteredService]:
"""
Discover services matching criteria with fallback support
"""
with self._lock:
candidates = [
s for s in self._services.values()
if s.name == name
and s.status in ("healthy", "unknown")
]
# Filter by environment (with fallback)
if environment:
env_candidates = [s for s in candidates if s.environment == environment]
if env_candidates:
candidates = env_candidates
# Filter by region (with fallback to any region)
if region:
region_candidates = [s for s in candidates if s.region == region]
if region_candidates:
candidates = region_candidates
# Filter by capabilities
if required_capabilities:
candidates = [
s for s in candidates
if required_capabilities.issubset(s.capabilities)
]
# Sort by environment priority and region proximity
candidates.sort(
key=lambda s: (
s.environment.value != Environment.PRODUCTION.value,
s.environment.value != Environment.STAGING.value,
s.region.value
)
)
return candidates
def update_health_status(self, service_id: str, status: str):
"""Update service health status"""
with self._lock:
if service_id in self._services:
self._services[service_id].status = status
self._services[service_id].last_health_check = time.time()
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
UPDATE services
SET status = ?, last_health_check = ?
WHERE service_id = ?
""", (status, time.time(), service_id))
def deregister_service(self, service_id: str) -> bool:
"""Remove service from registry"""
with self._lock:
if service_id in self._services:
del self._services[service_id]
with sqlite3.connect(self.db_path) as conn:
conn.execute("DELETE FROM services WHERE service_id = ?", (service_id,))
return True
return False
def list_all_services(self) -> List[Dict]:
"""List all registered services"""
with self._lock:
return [
{
"service_id": s.service_id,
"name": s.name,
"url": s.url,
"environment": s.environment.value,
"region": s.region.value,
"capabilities": list(s.capabilities),
"status": s.status,
"tags": s.tags
}
for s in self._services.values()
]
Import json for serialization
import json
async def demo_registry():
"""Demonstrate MCP Service Registry functionality"""
registry = MCPServiceRegistry(":memory:") # Use in-memory for demo
print("=== MCP Service Registry Demo ===\n")
# Register services across environments
services_to_register = [
{
"name": "vector-search",
"url": "https://vector-dev.internal/search",
"auth": {"type": "bearer", "token_env": "VECTOR_API_KEY"},
"env": Environment.DEVELOPMENT,
"region": Region.US_EAST,
"capabilities": ["embeddings", "similarity-search", " ANN"]
},
{
"name": "vector-search",
"url": "https://vector-prod.internal/search",
"auth": {"type": "bearer", "token_env": "VECTOR_API_KEY"},
"env": Environment.PRODUCTION,
"region": Region.US_EAST,
"capabilities": ["embeddings", "similarity-search", "ANN", "hybrid-search"]
},
{
"name": "document-store",
"url": "https://docs-staging.internal/api",
"auth": {"type": "api_key", "key_env": "DOCS_API_KEY"},
"env": Environment.STAGING,
"region": Region.EU_CENTRAL,
"capabilities": ["document-index", "full-text-search"]
},
{
"name": "document-store",
"url": "https://docs-prod.internal/api",
"auth": {"type": "api_key", "key_env": "DOCS_API_KEY"},
"env": Environment.PRODUCTION,
"region": Region.AP_SOUTHEAST,
"capabilities": ["document-index", "full-text-search", "semantic-search"]
}
]
print("Registering services...")
for svc in services_to_register:
service_id = registry.register_service(
name=svc["name"],
url=svc["url"],
auth_config=svc["auth"],
environment=svc["env"],
region=svc["region"],
capabilities=svc["capabilities"],
tags={"team": "platform", "owner": "infra-team"}
)
print(f" Registered: {svc['name']} ({svc['env'].value}) -> {service_id}")
# Mark as healthy for demo
registry.update_health_status(service_id, "healthy")
# Discover services with fallback
print("\n--- Discovering 'vector-search' service ---")
results = registry.discover_services(
"vector-search",
environment=Environment.PRODUCTION,
required_capabilities={"embeddings", "ANN"}
)
print(f"Found {len(results)} matching services (with fallback):")
for svc in results:
print(f" - {svc.name} ({svc.environment.value}/{svc.region.value}): {svc.url}")
print(f" Capabilities: {', '.join(svc.capabilities)}")
# Discover across all environments
print("\n--- Discovering all 'document-store' services ---")
results = registry.discover_services("document-store")
print(f"Found {len(results)} total services:")
for svc in results:
print(f" - {svc.name} ({svc.environment.value}/{svc.region.value})")
# List all services
print("\n--- All Registered Services ---")
all_services = registry.list_all_services()
for svc in all_services:
print(f" [{svc['service_id']}] {svc['name']} - {svc['status']}")
print(f" Env: {svc['environment']}, Region: {svc['region']}")
print(f" Capabilities: {svc['capabilities']}")
if __name__ == "__main__":
asyncio.run(demo_registry())Common Errors and Fixes
Based on extensive production experience debugging MCP integrations, here are the most common errors and their solutions:
Error 1: 401 Unauthorized on Service Discovery
# PROBLEM: Getting 401 errors when querying discovered services
ERROR: httpx.HTTPStatusError: 401 Client Error for url: https://.../mcp/health
ROOT CAUSE: The authentication token is missing, expired, or incorrect
SOLUTION: Ensure proper header construction for bearer tokens
WRONG:
headers = {"Authorization": api_key} # Missing "Bearer " prefix
CORRECT:
headers = {"Authorization": f"Bearer {self.api_key}"}
For HolySheep AI, always use:
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Get from https://www.holysheep.ai/register
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}Error 2: Connection Timeout in Service Discovery
# PROBLEM: Service discovery times out even though service is running
ERROR: httpx.TimeoutException: Connection timeout
ROOT CAUSE: Default timeout too short for slow services or high-latency networks
SOLUTION: Configure adaptive timeouts based on service priority
import httpx
class AdaptiveTimeoutClient:
"""Client with adaptive timeout based on service priority"""
@staticmethod
def get_timeout_for_service(service_priority: int) -> float:
"""
Calculate timeout based on service priority
Higher priority = shorter timeout
"""
base_timeout = 30.0 # seconds
if service_priority <= 1:
return 5.0 # Critical services: 5s timeout
elif service_priority <= 3:
return 10.0 # Important services: 10s timeout
elif service_priority <= 10:
return 30.0 # Standard services: 30s timeout
else:
return 60.0 # Low priority: 60s timeout
async def discover_with_adaptive_timeout(self, service: ServiceEndpoint):
"""Use service-specific timeout"""
timeout = self.get_timeout_for_service(service.priority)
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.get(
f"{service.url}/health",
headers=self._build_headers(service)
)
return response
Alternative: Use retry with exponential backoff
async def discover_with_retry(
service: ServiceEndpoint,
max_retries: int = 3,
base_delay: float = 1.0
):
"""Retry discovery with exponential backoff"""
import asyncio
for attempt in range(max_retries):
try:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(f"{service.url}/health")
return response.json()
except (httpx.TimeoutException, httpx.ConnectError) as e:
if attempt == max_retries - 1:
raise
delay = base_delay * (2 ** attempt)
print(f"Retry {attempt + 1}/{max_retries} after {delay}s")
await asyncio.sleep(delay)Error 3: Manifest Parsing Failures
# PROBLEM: Service discovery fails with JSONDecodeError or KeyError
ERROR: json.JSONDecodeError or KeyError: 'services'
ROOT CAUSE: Invalid manifest format or missing required fields
SOLUTION: Implement robust manifest validation with schema checking
from typing import Dict, Any, List
from pydantic import BaseModel, Field, validator
class ServiceConfig(BaseModel):
"""Validated service configuration"""
url: str = Field(..., description="Service endpoint URL")
auth: str = Field(default="bearer", description="Authentication type")
capabilities: List[str] = Field(default_factory=list)
priority: int = Field(default=100, ge=1, le=1000)
timeout_ms: int = Field(default=5000, ge=100, le=60000)
class MCPManifest(BaseModel):
"""Validated MCP manifest structure"""
version: str = Field(..., description="Manifest version")
services: Dict[str, ServiceConfig]
metadata: Dict[str, Any] = Field(default_factory=dict)
@validator('version')
def validate_version(cls, v):
if not v.startswith('1.'):
raise ValueError(f"Unsupported manifest version: {v}")
return v
def load_manifest_safely(manifest_data: Dict[str, Any]) -> MCPManifest:
"""Load and validate manifest with detailed error messages"""
try:
manifest = MCPManifest(**manifest_data)
return manifest
except Exception as e:
# Provide helpful error message
error_details = []
if hasattr(e, 'errors'):
for error in e.errors():
field = '.'.join(str(x) for x in error['loc'])
msg = error['msg']
error_details.append(f" - {field}: {msg}")
else:
error_details.append(f" - {str(e)}")
raise ValueError(
f"Invalid MCP manifest:\n" + "\n".join(error_details)
) from e
#