Async Patterns in MCP Python

Capsule description

The Python SDK for MCP is async-first. Every tool, resource, and prompt you define is an async function. This isn't a coincidence — it's a fundamental design decision. An MCP server that connects with external APIs, databases, or network services needs to handle multiple I/O operations without blocking. Async is what makes that possible.

If async/await in Python is familiar to you, this capsule reinforces your knowledge in the context of MCP. If async intimidates you, this capsule gives you the patterns you need to build robust MCP servers. You're not going to learn all of asyncio's theory — you're going to learn the specific patterns that real MCP servers use.

Why does async matter for MCP?

The problem: blocking I/O

Imagine an MCP server with a tool that queries an API:

# ❌ Sync: blocks the entire server during the request
import requests

@mcp.tool()
def get_weather(city: str) -> str:
    """Gets the weather for a city."""
    response = requests.get(f"https://api.weather.com/{city}")  # Blocks ~200ms
    return response.text

While that requests.get() waits for a response from the remote server (200ms, 500ms, sometimes seconds), your MCP server is frozen. It can't process other requests. If Claude Code sends another request, it has to wait.

# ✅ Async: the server can do other things while it waits
import httpx

@mcp.tool()
async def get_weather(city: str) -> str:
    """Gets the weather for a city."""
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.weather.com/{city}")  # Doesn't block
        return response.text

With await, the server says "I'm going to wait for this response, but in the meantime I can process other things." It's the difference between a bank teller (sync: one line, one customer at a time) and a restaurant (async: multiple tables served simultaneously).

When async makes a difference in MCP

*The operation's time is the same, but async lets the server do other things while it waits.

**When you have multiple tools doing I/O, async is the difference between a server that responds and one that hangs.

Async in MCP: the flow

Claude Code sends a request → MCP Server receives
                              ↓
                    Tool is an async function
                              ↓
                    await I/O operation (API, DB, file)
                              ↓
              [Server can process other requests while it waits]
                              ↓
                    I/O completes → Tool returns a result
                              ↓
                    MCP Server sends a response → Claude Code receives

Pattern 1: Async HTTP with httpx

Python's async HTTP client

httpx is the recommended library for async HTTP in Python. It's installed with the MCP SDK, so you already have it available.

Basic GET request

import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("http-examples")


@mcp.tool()
async def fetch_json(url: str) -> str:
    """Fetches JSON data from a 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 with a body

@mcp.tool()
async def post_data(url: str, payload: str) -> str:
    """Sends JSON data to a URL via POST."""
    import json

    try:
        data = json.loads(payload)
    except json.JSONDecodeError:
        return "Error: the payload isn't valid JSON."

    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 with headers and authentication

import os


@mcp.tool()
async def github_api(endpoint: str, method: str = "GET") -> str:
    """Queries the GitHub API.

    endpoint: relative path (e.g., '/repos/owner/repo').
    method: GET or POST.
    """
    token = os.environ.get("GITHUB_TOKEN")
    if not token:
        return "Error: GITHUB_TOKEN not configured."

    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: method {method} not supported."

        response.raise_for_status()
        return response.text[:5000]

Multiple concurrent requests

When you need data from multiple sources:

import asyncio
import json


@mcp.tool()
async def compare_apis(urls: list[str]) -> str:
    """Queries multiple URLs in parallel and compares the response times."""
    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() runs all the requests in parallel. If each request takes 200ms and you have 5 URLs, the total time is ~200ms, not 1000ms.

Pattern 2: Async Context Managers

What is an async context manager?

A context manager guarantees that a resource is opened and closed correctly. The async version does the same, but with async operations:

# Sync context manager (files)
with open("file.txt") as f:
    data = f.read()
# The file closes automatically

# Async context manager (HTTP connections)
async with httpx.AsyncClient() as client:
    response = await client.get(url)
# The connection closes automatically

Context manager for database connections

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:
    """Runs a read-only SQL query."""
    import json

    if not sql.strip().upper().startswith("SELECT"):
        return "Error: only SELECT queries are allowed."

    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)

Reusable context manager for APIs

from contextlib import asynccontextmanager


@asynccontextmanager
async def api_client(base_url: str, token: str | None = None):
    """Context manager for a configured API client."""
    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:
    """Lists a GitHub user's repositories."""
    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 lets you create async context managers using yield instead of defining classes with __aenter__ and __aexit__. More concise and Pythonic.

Pattern 3: Error handling in an async context

Try/except with async operations

Error handling in async is the same as in sync, but errors can come from more sources:

@mcp.tool()
async def robust_api_call(url: str) -> str:
    """Calls an API with complete error handling."""
    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"The API didn't respond in 10 seconds: {url}",
            "suggestion": "Verify that the URL is correct and that the service is available.",
        })

    except httpx.HTTPStatusError as e:
        return json.dumps({
            "error": "http_error",
            "status_code": e.response.status_code,
            "message": f"HTTP error {e.response.status_code} from {url}",
            "body": e.response.text[:500],
        })

    except httpx.ConnectError:
        return json.dumps({
            "error": "connection_error",
            "message": f"Could not connect to {url}",
            "suggestion": "Verify your network connection and that the host is accessible.",
        })

    except Exception as e:
        return json.dumps({
            "error": "unexpected",
            "type": type(e).__name__,
            "message": str(e),
        })

Retry with exponential backoff

import asyncio


async def retry_async(
    func,
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 30.0,
):
    """Runs an async function with retries and exponential backoff."""
    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:
    """Fetches data from a URL with automatic retries."""
    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"3 attempts to connect to {url} failed: {e}",
        })

Manual timeout with asyncio

@mcp.tool()
async def fetch_with_timeout(url: str, timeout_seconds: int = 30) -> str:
    """Fetches data with a configurable global timeout."""
    try:
        result = await asyncio.wait_for(
            _do_fetch(url),
            timeout=timeout_seconds,
        )
        return result
    except asyncio.TimeoutError:
        return f"Error: the complete operation exceeded {timeout_seconds} seconds."


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() is useful when you want a timeout for an entire compound operation, not just for the HTTP request.

Pattern 4: Async generators for large data

Data streaming

When a tool processes a lot of data, you can use async generators to process it incrementally:

async def read_large_file_lines(filepath: str):
    """Async generator that reads a file line by line."""
    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:
    """Analyzes a log file searching for a pattern.

    Processes the file line by line to handle large files.
    """
    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)

Process multiple files in parallel

@mcp.tool()
async def search_multiple_files(
    directory: str,
    pattern: str,
    file_extension: str = ".py",
) -> str:
    """Searches for a pattern in multiple files of a directory."""
    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)

Pattern 5: Shared state with async

The concurrency problem

When multiple tools access shared data, you need to protect the access:

import asyncio

mcp = FastMCP("concurrent-safe")

counter = {"value": 0}
lock = asyncio.Lock()


@mcp.tool()
async def increment_counter(amount: int = 1) -> str:
    """Increments the counter safely."""
    async with lock:
        counter["value"] += amount
        return f"Counter: {counter['value']}"


@mcp.tool()
async def get_counter() -> str:
    """Reads the counter's current value."""
    return f"Current value: {counter['value']}"

asyncio.Lock() guarantees that only one tool modifies the counter at a time. Without the lock, two simultaneous increments could produce incorrect results.

Async cache with 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:
    """Queries an API with a 10-minute cache."""
    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 result

Sync vs Async: when it matters in MCP

Practical rule

Does your tool do network I/O (API, database, remote service)?
├── Yes → ASYNC mandatory (httpx, aiofiles, aiosqlite)
└── No → Does your tool process large data?
    ├── Yes → ASYNC recommended (don't block the server)
    └── No → Async is the SDK's default, but a sync operation is fine inside async def

Sync operations inside async functions

Sometimes you need to call sync code from an async function. It's fine for fast operations:

@mcp.tool()
async def process_text(text: str) -> str:
    """Processes text — a fast CPU-bound operation."""
    words = text.split()           # sync, but instantaneous
    unique = set(words)            # sync, but instantaneous
    return f"Words: {len(words)}, Unique: {len(unique)}"

For sync operations that take time, use asyncio.to_thread():

import hashlib


def compute_hash_sync(data: str) -> str:
    """CPU-bound operation that takes time."""
    for _ in range(1000000):
        data = hashlib.sha256(data.encode()).hexdigest()
    return data


@mcp.tool()
async def heavy_computation(input_data: str) -> str:
    """Runs a heavy computation without blocking the server."""
    result = await asyncio.to_thread(compute_hash_sync, input_data)
    return f"Hash result: {result}"

asyncio.to_thread() runs the sync function in a separate thread, freeing the event loop.

Exercises

Exercise 1: Tool with multiple concurrent API calls (Medium)

Create a compare_weather tool that takes a list of cities and fetches the weather of all of them in parallel using asyncio.gather(). Use the public wttr.in API (no API key required).

import asyncio
import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("weather-compare")


@mcp.tool()
async def compare_weather(cities: list[str]) -> str:
    """Compares the weather of multiple cities in parallel."""
    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()

Exercise 2: Tool with retry and backoff (Medium)

Create a resilient_fetch tool that tries to fetch a URL up to 3 times, with exponential backoff (1s, 2s, 4s). Return the successful result or a summary of all the failed attempts.

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:
    """Fetches a URL with retries and exponential backoff."""
    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"All {max_retries} attempts failed for {url}",
    }, indent=2)


if __name__ == "__main__":
    mcp.run()

Exercise 3: Async cache with a resource (Medium)

Create an MCP server with:

• A tool cached_fetch that queries URLs with a 5-minute cache
• A resource cache://stats that shows cache statistics (hits, misses, items)
• A tool clear_cache that clears the cache

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:
    """Fetches a URL with a 5-minute cache."""
    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 - saved {int(time.monotonic() - timestamp)}s ago]\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 - fresh data]\n{result}"


@mcp.resource("cache://stats")
async def get_cache_stats() -> str:
    """Cache statistics."""
    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:
    """Clears the entire cache."""
    async with cache_lock:
        count = len(cache_data)
        cache_data.clear()
    return f"Cache cleared: {count} items removed."


if __name__ == "__main__":
    mcp.run()

Exercise 4: Reusable async context manager (Hard)

Create an async context manager APISession that:

• Configures an httpx.AsyncClient with a base_url and headers
• Automatically records the time of each request
• Keeps a log of the requests made
• Exposes the logs as a resource

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 that tracks requests automatically."""
    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:
    """Fetches info about a GitHub user."""
    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 of all the HTTP requests made."""
    return json.dumps({
        "total_requests": len(request_log),
        "logs": request_log[-20:],
    }, indent=2)


if __name__ == "__main__":
    mcp.run()

Exercise 5: Semaphore for rate limiting (Hard)

Create an MCP server that limits concurrency to 3 simultaneous requests using asyncio.Semaphore. If a 4th request arrives, it must wait for one of the 3 to finish.

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:
    """Fetches a URL respecting the limit of 3 concurrent requests."""
    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:
    """Rate limiter statistics."""
    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"

Cause: You're trying to use asyncio.run() inside a context that already has an event loop.

Solution:

# ❌ This fails if there's already an event loop
asyncio.run(some_coroutine())

# ✅ Inside an async function, use await directly
result = await some_coroutine()

# ✅ If you need to run from sync inside async, use nest_asyncio
import nest_asyncio
nest_asyncio.apply()

"httpx.ConnectTimeout when calling APIs"

Cause: The default timeout can be too short for slow APIs.

Solution:

async with httpx.AsyncClient(timeout=30) as client:
    response = await client.get(url)

"The shared data gets corrupted with multiple tools"

Cause: Concurrent access without protection.

Solution:

lock = asyncio.Lock()

@mcp.tool()
async def modify_data(value: str) -> str:
    async with lock:
        shared_data.append(value)
        return f"Added. Total: {len(shared_data)}"

Summary

In this capsule you learned:

• Async is mandatory in the Python SDK for MCP — all functions are async def
• httpx is the recommended async HTTP client (installed with the SDK)
• asyncio.gather() runs multiple I/O operations in parallel
• Async context managers guarantee cleanup of resources (connections, files)
• Retry with backoff is essential for production APIs
• asyncio.Lock() protects shared data between concurrent tools
• asyncio.Semaphore() limits concurrency for rate limiting
• asyncio.to_thread() runs heavy sync code without blocking the event loop

Next capsule: Project — a complete Python MCP server that connects with an external REST API. Everything you learned in capsules 02-04 converges here.

Additional resources

1. Python asyncio Documentation — Official reference
2. httpx — Async Support — Async HTTP client for Python
3. Real Python — Async IO — Complete asyncio tutorial
4. aiofiles — Async file I/O for Python
5. aiosqlite — Async SQLite for Python
6. MCP Python SDK — SDK reference

Next capsule: Project — build a complete Python MCP server that connects with a real REST API.