Módulo 5: MCP Server en Python
Patrones Async en MCP Python
Patrones Async en MCP Python
Descripción de la cápsula
El SDK de Python para MCP es async-first. Cada tool, resource, y prompt que defines es una función async. Esto no es una casualidad — es una decisión de diseño fundamental. Un MCP server que conecta con APIs externas, databases, o servicios de red necesita manejar múltiples operaciones de I/O sin bloquear. Async es lo que hace eso posible.
Si async/await en Python te resulta familiar, esta cápsula refuerza tu conocimiento en el contexto de MCP. Si async te intimida, esta cápsula te da los patrones que necesitas para construir MCP servers robustos. No vas a aprender toda la teoría de asyncio — vas a aprender los patrones específicos que usan los MCP servers reales.
¿Por qué async importa para MCP?
El problema: I/O blocking
Imagina un MCP server con un tool que consulta una API:
# ❌ Sync: bloquea el server entero durante la request
import requests
@mcp.tool()
def get_weather(city: str) -> str:
"""Obtiene el clima de una ciudad."""
response = requests.get(f"https://api.weather.com/{city}") # Bloquea ~200ms
return response.textMientras ese requests.get() espera respuesta del servidor remoto (200ms, 500ms, a veces segundos), tu MCP server está congelado. No puede procesar otros requests. Si Claude Code envía otro request, tiene que esperar.
# ✅ Async: el server puede hacer otras cosas mientras espera
import httpx
@mcp.tool()
async def get_weather(city: str) -> str:
"""Obtiene el clima de una ciudad."""
async with httpx.AsyncClient() as client:
response = await client.get(f"https://api.weather.com/{city}") # No bloquea
return response.textCon await, el server dice "voy a esperar esta respuesta, pero mientras tanto puedo procesar otras cosas." Es la diferencia entre un cajero de banco (sync: una fila, un cliente a la vez) y un restaurante (async: múltiples mesas atendidas simultáneamente).
Cuándo async marca la diferencia en MCP
| Operación | Sync | Async | Diferencia real |
|---|---|---|---|
| Leer un archivo local | ~1ms | ~1ms | Insignificante |
| Calcular un hash | ~5ms | ~5ms | Insignificante |
| Consultar una API REST | 100-2000ms | 100-2000ms* | Enorme** |
| Consultar una database | 10-500ms | 10-500ms* | Significativa |
| Descargar un archivo grande | 1-30s | 1-30s* | Enorme |
*El tiempo de la operación es el mismo, pero async permite que el server haga otras cosas mientras espera.
**Cuando tienes múltiples tools que hacen I/O, async es la diferencia entre un server que responde y uno que se cuelga.
Async en MCP: el flujo
Claude Code envía request → MCP Server recibe
↓
Tool es una función async
↓
await operación de I/O (API, DB, file)
↓
[Server puede procesar otros requests mientras espera]
↓
I/O completa → Tool retorna resultado
↓
MCP Server envía respuesta → Claude Code recibe
Patrón 1: Async HTTP con httpx
El cliente HTTP async de Python
httpx es la librería recomendada para HTTP async en Python. Se instala con el SDK de MCP, así que ya la tienes disponible.
GET request básico
import httpx
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("http-examples")
@mcp.tool()
async def fetch_json(url: str) -> str:
"""Obtiene datos JSON de una URL."""
import json
async with httpx.AsyncClient() as client:
response = await client.get(url, timeout=15)
response.raise_for_status()
data = response.json()
return json.dumps(data, indent=2)POST request con body
@mcp.tool()
async def post_data(url: str, payload: str) -> str:
"""Envía datos JSON a una URL via POST."""
import json
try:
data = json.loads(payload)
except json.JSONDecodeError:
return "Error: el payload no es JSON válido."
async with httpx.AsyncClient() as client:
response = await client.post(url, json=data, timeout=15)
return json.dumps(
{"status_code": response.status_code, "response": response.text[:2000]},
indent=2,
)Request con headers y autenticación
import os
@mcp.tool()
async def github_api(endpoint: str, method: str = "GET") -> str:
"""Consulta la API de GitHub.
endpoint: ruta relativa (e.g., '/repos/owner/repo').
method: GET o POST.
"""
token = os.environ.get("GITHUB_TOKEN")
if not token:
return "Error: GITHUB_TOKEN no configurado."
headers = {
"Authorization": f"Bearer {token}",
"Accept": "application/vnd.github.v3+json",
"X-GitHub-Api-Version": "2022-11-28",
}
base_url = "https://api.github.com"
url = f"{base_url}{endpoint}"
async with httpx.AsyncClient() as client:
if method.upper() == "GET":
response = await client.get(url, headers=headers, timeout=15)
elif method.upper() == "POST":
response = await client.post(url, headers=headers, timeout=15)
else:
return f"Error: método {method} no soportado."
response.raise_for_status()
return response.text[:5000]Múltiples requests concurrentes
Cuando necesitas datos de múltiples fuentes:
import asyncio
import json
@mcp.tool()
async def compare_apis(urls: list[str]) -> str:
"""Consulta múltiples URLs en paralelo y compara los tiempos de respuesta."""
import time
async def fetch_one(client: httpx.AsyncClient, url: str) -> dict:
start = time.monotonic()
try:
response = await client.get(url, timeout=10)
elapsed = time.monotonic() - start
return {
"url": url,
"status": response.status_code,
"time_ms": round(elapsed * 1000, 2),
"size_bytes": len(response.content),
}
except Exception as e:
elapsed = time.monotonic() - start
return {
"url": url,
"error": str(e),
"time_ms": round(elapsed * 1000, 2),
}
async with httpx.AsyncClient() as client:
tasks = [fetch_one(client, url) for url in urls]
results = await asyncio.gather(*tasks)
return json.dumps(results, indent=2)asyncio.gather() ejecuta todas las requests en paralelo. Si cada request toma 200ms y tienes 5 URLs, el tiempo total es ~200ms, no 1000ms.
Patrón 2: Async Context Managers
¿Qué es un async context manager?
Un context manager garantiza que un recurso se abra y se cierre correctamente. La versión async hace lo mismo, pero con operaciones async:
# Context manager sync (archivos)
with open("file.txt") as f:
data = f.read()
# El archivo se cierra automáticamente
# Context manager async (conexiones HTTP)
async with httpx.AsyncClient() as client:
response = await client.get(url)
# La conexión se cierra automáticamenteContext manager para conexiones de database
import aiosqlite
class DatabaseConnection:
def __init__(self, db_path: str):
self.db_path = db_path
self.db = None
async def __aenter__(self):
self.db = await aiosqlite.connect(self.db_path)
return self.db
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.db:
await self.db.close()
@mcp.tool()
async def query_database(sql: str) -> str:
"""Ejecuta una query SQL de solo lectura."""
import json
if not sql.strip().upper().startswith("SELECT"):
return "Error: solo queries SELECT están permitidas."
async with DatabaseConnection("data.db") as db:
db.row_factory = aiosqlite.Row
async with db.execute(sql) as cursor:
rows = await cursor.fetchall()
columns = [desc[0] for desc in cursor.description]
results = [dict(zip(columns, row)) for row in rows]
return json.dumps({"rows": len(results), "data": results}, indent=2)Context manager reutilizable para APIs
from contextlib import asynccontextmanager
@asynccontextmanager
async def api_client(base_url: str, token: str | None = None):
"""Context manager para un cliente API configurado."""
headers = {}
if token:
headers["Authorization"] = f"Bearer {token}"
async with httpx.AsyncClient(
base_url=base_url,
headers=headers,
timeout=15,
) as client:
yield client
@mcp.tool()
async def get_github_repos(username: str) -> str:
"""Lista repositorios de un usuario de GitHub."""
import json
token = os.environ.get("GITHUB_TOKEN")
async with api_client("https://api.github.com", token) as client:
response = await client.get(f"/users/{username}/repos")
response.raise_for_status()
repos = response.json()
summary = [
{"name": r["name"], "stars": r["stargazers_count"], "language": r["language"]}
for r in repos[:10]
]
return json.dumps(summary, indent=2)@asynccontextmanager te permite crear context managers async usando yield en lugar de definir clases con __aenter__ y __aexit__. Más conciso y Pythonic.
Patrón 3: Error handling en contexto async
Try/except con operaciones async
El error handling en async es igual que en sync, pero los errores pueden venir de más fuentes:
@mcp.tool()
async def robust_api_call(url: str) -> str:
"""Llama a una API con error handling completo."""
import json
try:
async with httpx.AsyncClient() as client:
response = await client.get(url, timeout=10)
response.raise_for_status()
return json.dumps(response.json(), indent=2)
except httpx.TimeoutException:
return json.dumps({
"error": "timeout",
"message": f"La API no respondió en 10 segundos: {url}",
"suggestion": "Verifica que la URL es correcta y que el servicio está disponible.",
})
except httpx.HTTPStatusError as e:
return json.dumps({
"error": "http_error",
"status_code": e.response.status_code,
"message": f"Error HTTP {e.response.status_code} de {url}",
"body": e.response.text[:500],
})
except httpx.ConnectError:
return json.dumps({
"error": "connection_error",
"message": f"No se pudo conectar a {url}",
"suggestion": "Verifica tu conexión de red y que el host es accesible.",
})
except Exception as e:
return json.dumps({
"error": "unexpected",
"type": type(e).__name__,
"message": str(e),
})Retry con backoff exponencial
import asyncio
async def retry_async(
func,
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 30.0,
):
"""Ejecuta una función async con reintentos y backoff exponencial."""
last_error = None
for attempt in range(max_retries):
try:
return await func()
except (httpx.TimeoutException, httpx.ConnectError) as e:
last_error = e
if attempt < max_retries - 1:
delay = min(base_delay * (2 ** attempt), max_delay)
await asyncio.sleep(delay)
raise last_error
@mcp.tool()
async def reliable_fetch(url: str) -> str:
"""Obtiene datos de una URL con reintentos automáticos."""
import json
async def do_fetch():
async with httpx.AsyncClient() as client:
response = await client.get(url, timeout=10)
response.raise_for_status()
return response.text
try:
result = await retry_async(do_fetch, max_retries=3)
return result[:5000]
except Exception as e:
return json.dumps({
"error": "all_retries_failed",
"message": f"Fallaron 3 intentos de conectar a {url}: {e}",
})Timeout manual con asyncio
@mcp.tool()
async def fetch_with_timeout(url: str, timeout_seconds: int = 30) -> str:
"""Obtiene datos con un timeout global configurable."""
try:
result = await asyncio.wait_for(
_do_fetch(url),
timeout=timeout_seconds,
)
return result
except asyncio.TimeoutError:
return f"Error: la operación completa excedió {timeout_seconds} segundos."
async def _do_fetch(url: str) -> str:
async with httpx.AsyncClient() as client:
response = await client.get(url, timeout=None)
response.raise_for_status()
return response.text[:5000]asyncio.wait_for() es útil cuando quieres un timeout para toda una operación compuesta, no solo para el request HTTP.
Patrón 4: Async generators para datos grandes
Streaming de datos
Cuando un tool procesa muchos datos, puedes usar async generators para procesarlos incrementalmente:
async def read_large_file_lines(filepath: str):
"""Async generator que lee un archivo línea por línea."""
import aiofiles
async with aiofiles.open(filepath, mode="r") as f:
async for line in f:
yield line
@mcp.tool()
async def analyze_log_file(filepath: str, pattern: str) -> str:
"""Analiza un archivo de log buscando un patrón.
Procesa el archivo línea por línea para manejar archivos grandes.
"""
import json
import re
matches = []
line_count = 0
async for line in read_large_file_lines(filepath):
line_count += 1
if re.search(pattern, line):
matches.append({
"line_number": line_count,
"content": line.strip()[:200],
})
if len(matches) >= 100:
break
return json.dumps({
"file": filepath,
"pattern": pattern,
"lines_scanned": line_count,
"matches_found": len(matches),
"matches": matches,
}, indent=2)Procesar múltiples archivos en paralelo
@mcp.tool()
async def search_multiple_files(
directory: str,
pattern: str,
file_extension: str = ".py",
) -> str:
"""Busca un patrón en múltiples archivos de un directorio."""
import json
import re
files_to_search = []
for root, _dirs, files in os.walk(directory):
for fname in files:
if fname.endswith(file_extension):
files_to_search.append(os.path.join(root, fname))
async def search_file(filepath: str) -> dict:
try:
with open(filepath, "r", encoding="utf-8", errors="ignore") as f:
content = f.read()
finds = [
{"line": i + 1, "text": line.strip()[:150]}
for i, line in enumerate(content.splitlines())
if re.search(pattern, line)
]
return {"file": filepath, "matches": len(finds), "details": finds[:10]}
except Exception as e:
return {"file": filepath, "error": str(e)}
tasks = [search_file(f) for f in files_to_search[:50]]
results = await asyncio.gather(*tasks)
results_with_matches = [r for r in results if r.get("matches", 0) > 0]
return json.dumps({
"directory": directory,
"pattern": pattern,
"files_searched": len(files_to_search),
"files_with_matches": len(results_with_matches),
"results": results_with_matches,
}, indent=2)Patrón 5: Estado compartido con async
El problema de la concurrencia
Cuando múltiples tools acceden a datos compartidos, necesitas proteger el acceso:
import asyncio
mcp = FastMCP("concurrent-safe")
counter = {"value": 0}
lock = asyncio.Lock()
@mcp.tool()
async def increment_counter(amount: int = 1) -> str:
"""Incrementa el contador de forma segura."""
async with lock:
counter["value"] += amount
return f"Contador: {counter['value']}"
@mcp.tool()
async def get_counter() -> str:
"""Lee el valor actual del contador."""
return f"Valor actual: {counter['value']}"asyncio.Lock() garantiza que solo un tool modifica el contador a la vez. Sin el lock, dos incrementos simultáneos podrían producir resultados incorrectos.
Cache async con TTL
import time
class AsyncCache:
def __init__(self, ttl_seconds: int = 300):
self._cache: dict[str, tuple[float, str]] = {}
self._ttl = ttl_seconds
self._lock = asyncio.Lock()
async def get(self, key: str) -> str | None:
async with self._lock:
if key in self._cache:
timestamp, value = self._cache[key]
if time.monotonic() - timestamp < self._ttl:
return value
del self._cache[key]
return None
async def set(self, key: str, value: str):
async with self._lock:
self._cache[key] = (time.monotonic(), value)
cache = AsyncCache(ttl_seconds=600)
@mcp.tool()
async def cached_api_call(url: str) -> str:
"""Consulta una API con cache de 10 minutos."""
cached = await cache.get(url)
if cached:
return f"[CACHE HIT]\n{cached}"
async with httpx.AsyncClient() as client:
response = await client.get(url, timeout=10)
response.raise_for_status()
result = response.text[:5000]
await cache.set(url, result)
return resultSync vs Async: cuándo importa en MCP
Regla práctica
¿Tu tool hace I/O de red (API, database, servicio remoto)?
├── Sí → ASYNC obligatorio (httpx, aiofiles, aiosqlite)
└── No → ¿Tu tool procesa datos grandes?
├── Sí → ASYNC recomendado (no bloquear el server)
└── No → Async es el default del SDK, pero la operación sync está bien dentro de async def
Operaciones sync dentro de funciones async
A veces necesitas llamar código sync desde una función async. Está bien para operaciones rápidas:
@mcp.tool()
async def process_text(text: str) -> str:
"""Procesa texto — operación CPU-bound rápida."""
words = text.split() # sync, pero instantáneo
unique = set(words) # sync, pero instantáneo
return f"Palabras: {len(words)}, Únicas: {len(unique)}"Para operaciones sync que toman tiempo, usa asyncio.to_thread():
import hashlib
def compute_hash_sync(data: str) -> str:
"""Operación CPU-bound que toma tiempo."""
for _ in range(1000000):
data = hashlib.sha256(data.encode()).hexdigest()
return data
@mcp.tool()
async def heavy_computation(input_data: str) -> str:
"""Ejecuta un cálculo pesado sin bloquear el server."""
result = await asyncio.to_thread(compute_hash_sync, input_data)
return f"Hash resultado: {result}"asyncio.to_thread() ejecuta la función sync en un thread separado, liberando el event loop.
Ejercicios
Ejercicio 1: Tool con múltiples API calls concurrentes (Medio)
Crea un tool compare_weather que reciba una lista de ciudades y obtenga el clima de todas en paralelo usando asyncio.gather(). Usa la API pública de wttr.in (no requiere API key).
Ver solución
import asyncio
import httpx
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("weather-compare")
@mcp.tool()
async def compare_weather(cities: list[str]) -> str:
"""Compara el clima de múltiples ciudades en paralelo."""
import json
async def get_city_weather(client: httpx.AsyncClient, city: str) -> dict:
try:
response = await client.get(
f"https://wttr.in/{city}?format=j1",
timeout=10,
)
response.raise_for_status()
data = response.json()
current = data["current_condition"][0]
return {
"city": city,
"temp_c": current["temp_C"],
"feels_like_c": current["FeelsLikeC"],
"description": current["weatherDesc"][0]["value"],
"humidity": current["humidity"],
"wind_kmh": current["windspeedKmph"],
}
except Exception as e:
return {"city": city, "error": str(e)}
async with httpx.AsyncClient() as client:
tasks = [get_city_weather(client, city) for city in cities]
results = await asyncio.gather(*tasks)
return json.dumps(results, indent=2, ensure_ascii=False)
if __name__ == "__main__":
mcp.run()Ejercicio 2: Tool con retry y backoff (Medio)
Crea un tool resilient_fetch que intente obtener una URL hasta 3 veces, con backoff exponencial (1s, 2s, 4s). Retorna el resultado exitoso o un resumen de todos los intentos fallidos.
Ver solución
import asyncio
import httpx
import json
import time
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("resilient-client")
@mcp.tool()
async def resilient_fetch(url: str, max_retries: int = 3) -> str:
"""Obtiene una URL con reintentos y backoff exponencial."""
attempts = []
async with httpx.AsyncClient() as client:
for attempt in range(max_retries):
start = time.monotonic()
try:
response = await client.get(url, timeout=10)
response.raise_for_status()
elapsed = time.monotonic() - start
return json.dumps({
"success": True,
"attempt": attempt + 1,
"time_ms": round(elapsed * 1000, 2),
"status_code": response.status_code,
"data": response.text[:3000],
}, indent=2)
except Exception as e:
elapsed = time.monotonic() - start
attempts.append({
"attempt": attempt + 1,
"error": str(e),
"time_ms": round(elapsed * 1000, 2),
})
if attempt < max_retries - 1:
delay = 2 ** attempt
await asyncio.sleep(delay)
return json.dumps({
"success": False,
"total_attempts": max_retries,
"attempts": attempts,
"message": f"Todos los {max_retries} intentos fallaron para {url}",
}, indent=2)
if __name__ == "__main__":
mcp.run()Ejercicio 3: Cache async con resource (Medio)
Crea un MCP server con:
- Un tool
cached_fetchque consulte URLs con cache de 5 minutos - Un resource
cache://statsque muestre estadísticas del cache (hits, misses, items) - Un tool
clear_cacheque limpie el cache
Ver solución
import asyncio
import time
import httpx
import json
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("cached-fetcher")
cache_data: dict[str, tuple[float, str]] = {}
cache_stats = {"hits": 0, "misses": 0}
TTL = 300
cache_lock = asyncio.Lock()
@mcp.tool()
async def cached_fetch(url: str) -> str:
"""Obtiene una URL con cache de 5 minutos."""
async with cache_lock:
if url in cache_data:
timestamp, value = cache_data[url]
if time.monotonic() - timestamp < TTL:
cache_stats["hits"] += 1
return f"[CACHE HIT - guardado hace {int(time.monotonic() - timestamp)}s]\n{value}"
del cache_data[url]
cache_stats["misses"] += 1
async with httpx.AsyncClient() as client:
response = await client.get(url, timeout=10)
response.raise_for_status()
result = response.text[:3000]
async with cache_lock:
cache_data[url] = (time.monotonic(), result)
return f"[CACHE MISS - datos frescos]\n{result}"
@mcp.resource("cache://stats")
async def get_cache_stats() -> str:
"""Estadísticas del cache."""
total = cache_stats["hits"] + cache_stats["misses"]
hit_rate = (cache_stats["hits"] / total * 100) if total > 0 else 0
return json.dumps({
"hits": cache_stats["hits"],
"misses": cache_stats["misses"],
"total_requests": total,
"hit_rate_percent": round(hit_rate, 1),
"cached_items": len(cache_data),
}, indent=2)
@mcp.tool()
async def clear_cache() -> str:
"""Limpia todo el cache."""
async with cache_lock:
count = len(cache_data)
cache_data.clear()
return f"Cache limpiado: {count} items eliminados."
if __name__ == "__main__":
mcp.run()Ejercicio 4: Async context manager reutilizable (Difícil)
Crea un context manager async APISession que:
- Configure un httpx.AsyncClient con base_url y headers
- Registre automáticamente el tiempo de cada request
- Lleve un log de requests realizados
- Expón los logs como un resource
Ver solución
import asyncio
import time
import httpx
import json
from contextlib import asynccontextmanager
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("api-session-demo")
request_log: list[dict] = []
log_lock = asyncio.Lock()
@asynccontextmanager
async def api_session(base_url: str, headers: dict | None = None):
"""Context manager que trackea requests automáticamente."""
client = httpx.AsyncClient(
base_url=base_url,
headers=headers or {},
timeout=15,
)
original_get = client.get
original_post = client.post
async def tracked_get(url, **kwargs):
start = time.monotonic()
response = await original_get(url, **kwargs)
elapsed = time.monotonic() - start
async with log_lock:
request_log.append({
"method": "GET",
"url": f"{base_url}{url}",
"status": response.status_code,
"time_ms": round(elapsed * 1000, 2),
"timestamp": time.strftime("%H:%M:%S"),
})
return response
async def tracked_post(url, **kwargs):
start = time.monotonic()
response = await original_post(url, **kwargs)
elapsed = time.monotonic() - start
async with log_lock:
request_log.append({
"method": "POST",
"url": f"{base_url}{url}",
"status": response.status_code,
"time_ms": round(elapsed * 1000, 2),
"timestamp": time.strftime("%H:%M:%S"),
})
return response
client.get = tracked_get
client.post = tracked_post
try:
yield client
finally:
await client.aclose()
@mcp.tool()
async def fetch_github_user(username: str) -> str:
"""Obtiene info de un usuario de GitHub."""
async with api_session("https://api.github.com") as client:
response = await client.get(f"/users/{username}")
response.raise_for_status()
data = response.json()
return json.dumps({
"login": data["login"],
"name": data.get("name"),
"repos": data["public_repos"],
"followers": data["followers"],
}, indent=2)
@mcp.resource("logs://requests")
async def get_request_logs() -> str:
"""Log de todos los requests HTTP realizados."""
return json.dumps({
"total_requests": len(request_log),
"logs": request_log[-20:],
}, indent=2)
if __name__ == "__main__":
mcp.run()Ejercicio 5: Semáforo para rate limiting (Difícil)
Crea un MCP server que limite la concurrencia a 3 requests simultáneos usando asyncio.Semaphore. Si un 4to request llega, debe esperar a que uno de los 3 termine.
Ver solución
import asyncio
import httpx
import json
import time
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("rate-limited")
semaphore = asyncio.Semaphore(3)
active_requests = {"count": 0}
stats = {"total": 0, "queued": 0}
@mcp.tool()
async def rate_limited_fetch(url: str) -> str:
"""Obtiene una URL respetando el límite de 3 requests concurrentes."""
queued = semaphore.locked()
if queued:
stats["queued"] += 1
start_wait = time.monotonic()
async with semaphore:
wait_time = time.monotonic() - start_wait
active_requests["count"] += 1
stats["total"] += 1
current_active = active_requests["count"]
try:
async with httpx.AsyncClient() as client:
start = time.monotonic()
response = await client.get(url, timeout=15)
response.raise_for_status()
fetch_time = time.monotonic() - start
return json.dumps({
"url": url,
"status": response.status_code,
"fetch_time_ms": round(fetch_time * 1000, 2),
"wait_time_ms": round(wait_time * 1000, 2),
"was_queued": queued,
"concurrent_requests": current_active,
"data_preview": response.text[:500],
}, indent=2)
finally:
active_requests["count"] -= 1
@mcp.resource("ratelimit://stats")
async def get_rate_limit_stats() -> str:
"""Estadísticas del rate limiter."""
return json.dumps({
"max_concurrent": 3,
"currently_active": active_requests["count"],
"total_requests": stats["total"],
"times_queued": stats["queued"],
}, indent=2)
if __name__ == "__main__":
mcp.run()Troubleshooting
"RuntimeError: This event loop is already running"
Causa: Intentas usar asyncio.run() dentro de un contexto que ya tiene event loop.
Solución:
# ❌ Esto falla si ya hay event loop
asyncio.run(some_coroutine())
# ✅ Dentro de una función async, usa await directamente
result = await some_coroutine()
# ✅ Si necesitas correr desde sync dentro de async, usa nest_asyncio
import nest_asyncio
nest_asyncio.apply()"httpx.ConnectTimeout al llamar APIs"
Causa: El timeout default puede ser muy corto para APIs lentas.
Solución:
async with httpx.AsyncClient(timeout=30) as client:
response = await client.get(url)"Los datos compartidos se corrompen con múltiples tools"
Causa: Acceso concurrente sin protección.
Solución:
lock = asyncio.Lock()
@mcp.tool()
async def modify_data(value: str) -> str:
async with lock:
shared_data.append(value)
return f"Agregado. Total: {len(shared_data)}"Resumen
En esta cápsula aprendiste:
- Async es obligatorio en el SDK de Python para MCP — todas las funciones son
async def - httpx es el cliente HTTP async recomendado (instalado con el SDK)
asyncio.gather()ejecuta múltiples operaciones de I/O en paralelo- Async context managers garantizan limpieza de recursos (conexiones, archivos)
- Retry con backoff es esencial para APIs de producción
asyncio.Lock()protege datos compartidos entre tools concurrentesasyncio.Semaphore()limita la concurrencia para rate limitingasyncio.to_thread()ejecuta código sync pesado sin bloquear el event loop
Próxima cápsula: Proyecto — MCP server Python completo que conecta con una API REST externa. Todo lo que aprendiste en las cápsulas 02-04 converge aquí.
Recursos adicionales
- Python asyncio Documentation — Referencia oficial
- httpx — Async Support — Cliente HTTP async para Python
- Real Python — Async IO — Tutorial completo de asyncio
- aiofiles — File I/O async para Python
- aiosqlite — SQLite async para Python
- MCP Python SDK — Referencia del SDK
Siguiente cápsula: Proyecto — construye un MCP server Python completo que conecta con una API REST real.