Implementing Async Tool Functions in FastMCP: A Complete Guide
FastMCP enables seamless registration of asynchronous Python functions as callable tools using the @mcp.tool decorator, allowing non-blocking execution while maintaining type safety through TypedDict schemas.
The cr2007/mcp-wordle-python repository demonstrates how to build a Model Context Protocol (MCP) server that exposes asynchronous tools to language model agents. This guide examines how to implement async tool functions in FastMCP, from schema definition to server execution.
Understanding FastMCP Architecture
Initializing the FastMCP Instance
The foundation of any MCP server is the FastMCP class, which acts as a registry for tools and handles protocol communication. In src/mcp_wordle/main.py, the server initializes with a descriptive name:
from fastmcp import FastMCP
mcp = FastMCP("WordleMCP")
Defining Async Tool Functions in FastMCP
Creating TypedDict Schemas for Type Safety
Before implementing the tool logic, define the data structures using TypedDict to provide JSON schema metadata. The Wordle implementation defines two schemas in src/mcp_wordle/main.py:
from typing import TypedDict, Union
from datetime import date
class WordleAPIData(TypedDict):
id: int
solution: str
print_date: str
days_since_launch: int
editor: str
class WordleError(TypedDict):
error: str
status_code: int
Registering Tools with the @mcp.tool Decorator
FastMCP converts Python functions into protocol-compatible tools using the @mcp.tool decorator. The decorator accepts parameters that map to OpenAI-compatible tool definitions:
@mcp.tool(
name="get_wordle_solution",
description=(
"Fetches the Wordle of a particular date provided "
"between 2021-05-19 to 23 days future"
),
annotations={"readOnlyHint": True},
)
async def get_wordle_data(
target_date: str = date.today().isoformat()
) -> Union[WordleAPIData, WordleError]:
# Implementation details below
pass
Implementing the Async Wordle Tool
Building the HTTP Request Logic
The tool implementation constructs the API endpoint and executes the HTTP request. While the current implementation uses the synchronous requests library, the async def signature ensures compatibility with async event loops:
import requests
@mcp.tool(
name="get_wordle_solution",
description="Fetches the Wordle solution for a given date.",
annotations={"readOnlyHint": True},
)
async def get_wordle_data(
target_date: str = date.today().isoformat()
) -> Union[WordleAPIData, WordleError]:
url = f"https://www.nytimes.com/svc/wordle/v2/{target_date}.json"
return requests.get(url, timeout=300).json()
Handling Async Execution
The async def declaration allows FastMCP to execute the tool without blocking the main event loop. This is crucial for MCP servers handling multiple concurrent tool calls from language model agents. While the example uses requests, production implementations should migrate to httpx.AsyncClient for true asynchronous I/O:
import httpx
async def get_wordle_data(
target_date: str = date.today().isoformat()
) -> Union[WordleAPIData, WordleError]:
url = f"https://www.nytimes.com/svc/wordle/v2/{target_date}.json"
async with httpx.AsyncClient() as client:
response = await client.get(url, timeout=30.0)
return response.json()
Running the FastMCP Server
Once tools are registered, the server starts via the mcp.run() method. In src/mcp_wordle/main.py, the entry point checks for direct execution:
if __name__ == "__main__":
mcp.run()
By default, FastMCP serves on http://localhost:8000, exposing registered tools at the /tools/{tool_name} endpoint. Clients can invoke the async Wordle tool via HTTP POST requests:
import httpx
response = httpx.post(
"http://localhost:8000/tools/get_wordle_solution",
json={"target_date": "2024-02-20"}
)
print(response.json())
Summary
- FastMCP initialization requires creating a named
FastMCPinstance that acts as the tool registry. - TypedDict schemas provide JSON schema metadata for tool inputs and outputs, enabling type-safe interactions with language model agents.
- The
@mcp.tooldecorator registers async functions as callable tools, acceptingname,description, andannotationsparameters for OpenAI-compatible definitions. - Async function signatures (
async def) ensure non-blocking execution within the MCP server, supporting concurrent tool invocations. - Server execution via
mcp.run()exposes tools over HTTP, allowing external clients to invoke async functions remotely.
Frequently Asked Questions
What is FastMCP and how does it handle async tools?
FastMCP is a lightweight Python framework that wraps functions into Model Context Protocol (MCP) compatible tools. It handles async tools by detecting async def signatures and executing them within the server's event loop without blocking other operations. According to the cr2007/mcp-wordle-python source code, this allows the server to maintain responsiveness while waiting for external API responses.
How do I register an async function as a FastMCP tool?
Register an async function using the @mcp.tool() decorator with explicit metadata parameters. As shown in src/mcp_wordle/main.py, you must provide the name, description, and optional annotations (such as readOnlyHint), then define the function with async def and appropriate type hints. FastMCP automatically generates the JSON schema from your TypedDict return types and parameter annotations.
Can I use synchronous libraries like requests in async FastMCP tools?
While FastMCP supports async def functions, using synchronous I/O libraries like requests inside them blocks the event loop. The cr2007/mcp-wordle-python implementation currently uses requests.get() within an async function, which works but isn't optimal. For production async tools, migrate to httpx.AsyncClient or aiohttp to achieve true non-blocking I/O while maintaining the async def signature required by FastMCP.
How do I run the FastMCP server after implementing async tools?
Execute the server by calling mcp.run() within a if __name__ == "__main__": block, as implemented in src/mcp_wordle/main.py. This starts the default ASGI server on http://localhost:8000, exposing your async tools at /tools/{tool_name} endpoints. Clients can then invoke your async functions via HTTP POST requests with JSON payloads matching your defined parameter schemas.
Have a question about this repo?
These articles cover the highlights, but your codebase questions are specific. Give your agent direct access to the source. Share this with your agent to get started:
curl -s "https://instagit.com/install.md" Maintain an open-source project? Get it listed too →