SDK de Python para MCP
Implementación en Python del Protocolo de Contexto de Modelo (MCP)
Tabla de contenido
Descripción general
El Protocolo de Contexto de Modelo permite que las aplicaciones proporcionen contexto para los LLM de forma estandarizada, separando la necesidad de proporcionar contexto de la interacción real con el LLM. Este SDK de Python implementa la especificación MCP completa, lo que facilita:
- Cree clientes MCP que puedan conectarse a cualquier servidor MCP
- Cree servidores MCP que expongan recursos, indicaciones y herramientas
- Utilice transportes estándar como stdio y SSE
- Manejar todos los mensajes del protocolo MCP y eventos del ciclo de vida
Instalación
Cómo agregar MCP a su proyecto de Python
Recomendamos utilizar uv para administrar sus proyectos de Python.
Si aún no ha creado un proyecto administrado por uv, cree uno:
uv init mcp-server-demo
cd mcp-server-demo
Luego agregue MCP a las dependencias de su proyecto:
Alternativamente, para proyectos que utilizan pip para dependencias:
Ejecución de las herramientas de desarrollo de MCP independientes
Para ejecutar el comando mcp con uv:
Inicio rápido
Creemos un servidor MCP simple que exponga una herramienta de calculadora y algunos datos:
# server.py
from mcp.server.fastmcp import FastMCP
# Create an MCP server
mcp = FastMCP("Demo")
# Add an addition tool
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b
# Add a dynamic greeting resource
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
"""Get a personalized greeting"""
return f"Hello, {name}!"
Puede instalar este servidor en Claude Desktop e interactuar con él de inmediato ejecutando:
Alternativamente, puedes probarlo con el Inspector MCP:
¿Qué es MCP?
El Protocolo de Contexto de Modelo (MCP) permite crear servidores que exponen datos y funcionalidades a aplicaciones LLM de forma segura y estandarizada. Se trata de una API web, pero diseñada específicamente para interacciones LLM. Los servidores MCP pueden:
- Exponer datos a través de recursos (piense en ellos como puntos finales GET; se utilizan para cargar información en el contexto del LLM)
- Proporcionar funcionalidad a través de herramientas (algo así como puntos finales POST; se utilizan para ejecutar código o producir un efecto secundario)
- Definir patrones de interacción a través de Prompts (plantillas reutilizables para interacciones LLM)
- ¡Y más!
Conceptos básicos
Servidor
El servidor FastMCP es la interfaz principal del protocolo MCP. Se encarga de la gestión de conexiones, la conformidad con el protocolo y el enrutamiento de mensajes.
# Add lifespan support for startup/shutdown with strong typing
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from dataclasses import dataclass
from fake_database import Database # Replace with your actual DB type
from mcp.server.fastmcp import Context, FastMCP
# Create a named server
mcp = FastMCP("My App")
# Specify dependencies for deployment and development
mcp = FastMCP("My App", dependencies=["pandas", "numpy"])
@dataclass
class AppContext:
db: Database
@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
"""Manage application lifecycle with type-safe context"""
# Initialize on startup
db = await Database.connect()
try:
yield AppContext(db=db)
finally:
# Cleanup on shutdown
await db.disconnect()
# Pass lifespan to server
mcp = FastMCP("My App", lifespan=app_lifespan)
# Access type-safe lifespan context in tools
@mcp.tool()
def query_db(ctx: Context) -> str:
"""Tool that uses initialized resources"""
db = ctx.request_context.lifespan_context["db"]
return db.query()
Recursos
Los recursos son la forma de exponer los datos a los LLM. Son similares a los puntos finales GET de una API REST: proporcionan datos, pero no deberían realizar cálculos significativos ni tener efectos secundarios.
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("My App")
@mcp.resource("config://app")
def get_config() -> str:
"""Static configuration data"""
return "App configuration here"
@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: str) -> str:
"""Dynamic user data"""
return f"Profile data for user {user_id}"
Herramientas
Las herramientas permiten a los LLM realizar acciones a través de su servidor. A diferencia de los recursos, se espera que las herramientas realicen cálculos y tengan efectos secundarios:
import httpx
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("My App")
@mcp.tool()
def calculate_bmi(weight_kg: float, height_m: float) -> float:
"""Calculate BMI given weight in kg and height in meters"""
return weight_kg / (height_m**2)
@mcp.tool()
async def fetch_weather(city: str) -> str:
"""Fetch current weather for a city"""
async with httpx.AsyncClient() as client:
response = await client.get(f"https://api.weather.com/{city}")
return response.text
Indicaciones
Las indicaciones son plantillas reutilizables que ayudan a los LLM a interactuar con su servidor de manera efectiva:
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.prompts import base
mcp = FastMCP("My App")
@mcp.prompt()
def review_code(code: str) -> str:
return f"Please review this code:\n\n{code}"
@mcp.prompt()
def debug_error(error: str) -> list[base.Message]:
return [
base.UserMessage("I'm seeing this error:"),
base.UserMessage(error),
base.AssistantMessage("I'll help debug that. What have you tried so far?"),
]
Imágenes
FastMCP proporciona una clase Image que maneja automáticamente los datos de imagen:
from mcp.server.fastmcp import FastMCP, Image
from PIL import Image as PILImage
mcp = FastMCP("My App")
@mcp.tool()
def create_thumbnail(image_path: str) -> Image:
"""Create a thumbnail from an image"""
img = PILImage.open(image_path)
img.thumbnail((100, 100))
return Image(data=img.tobytes(), format="png")
Contexto
El objeto Contexto brinda a sus herramientas y recursos acceso a las capacidades de MCP:
from mcp.server.fastmcp import FastMCP, Context
mcp = FastMCP("My App")
@mcp.tool()
async def long_task(files: list[str], ctx: Context) -> str:
"""Process multiple files with progress tracking"""
for i, file in enumerate(files):
ctx.info(f"Processing {file}")
await ctx.report_progress(i, len(files))
data, mime_type = await ctx.read_resource(f"file://{file}")
return "Processing complete"
Ejecución de su servidor
Modo de desarrollo
La forma más rápida de probar y depurar su servidor es con el Inspector MCP:
mcp dev server.py
# Add dependencies
mcp dev server.py --with pandas --with numpy
# Mount local code
mcp dev server.py --with-editable .
Integración de escritorio de Claude
Una vez que su servidor esté listo, instálelo en Claude Desktop:
mcp install server.py
# Custom name
mcp install server.py --name "My Analytics Server"
# Environment variables
mcp install server.py -v API_KEY=abc123 -v DB_URL=postgres://...
mcp install server.py -f .env
Ejecución directa
Para escenarios avanzados como implementaciones personalizadas:
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("My App")
if __name__ == "__main__":
mcp.run()
Ejecútalo con:
python server.py
# or
mcp run server.py
Montaje en un servidor ASGI existente
Puede montar el servidor SSE en un servidor ASGI existente mediante el método sse_app . Esto le permite integrar el servidor SSE con otras aplicaciones ASGI.
from starlette.applications import Starlette
from starlette.routing import Mount, Host
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("My App")
# Mount the SSE server to the existing ASGI server
app = Starlette(
routes=[
Mount('/', app=mcp.sse_app()),
]
)
# or dynamically mount as host
app.router.routes.append(Host('mcp.acme.corp', app=mcp.sse_app()))
Para obtener más información sobre el montaje de aplicaciones en Starlette, consulte la documentación de Starlette .
Ejemplos
Servidor Echo
Un servidor sencillo que muestra recursos, herramientas y sugerencias:
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Echo")
@mcp.resource("echo://{message}")
def echo_resource(message: str) -> str:
"""Echo a message as a resource"""
return f"Resource echo: {message}"
@mcp.tool()
def echo_tool(message: str) -> str:
"""Echo a message as a tool"""
return f"Tool echo: {message}"
@mcp.prompt()
def echo_prompt(message: str) -> str:
"""Create an echo prompt"""
return f"Please process this message: {message}"
Explorador de SQLite
Un ejemplo más complejo que muestra la integración de bases de datos:
import sqlite3
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("SQLite Explorer")
@mcp.resource("schema://main")
def get_schema() -> str:
"""Provide the database schema as a resource"""
conn = sqlite3.connect("database.db")
schema = conn.execute("SELECT sql FROM sqlite_master WHERE type='table'").fetchall()
return "\n".join(sql[0] for sql in schema if sql[0])
@mcp.tool()
def query_data(sql: str) -> str:
"""Execute SQL queries safely"""
conn = sqlite3.connect("database.db")
try:
result = conn.execute(sql).fetchall()
return "\n".join(str(row) for row in result)
except Exception as e:
return f"Error: {str(e)}"
Uso avanzado
Servidor de bajo nivel
Para mayor control, puede usar directamente la implementación del servidor de bajo nivel. Esto le brinda acceso completo al protocolo y le permite personalizar cada aspecto de su servidor, incluyendo la gestión del ciclo de vida mediante la API de vida útil.
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from fake_database import Database # Replace with your actual DB type
from mcp.server import Server
@asynccontextmanager
async def server_lifespan(server: Server) -> AsyncIterator[dict]:
"""Manage server startup and shutdown lifecycle."""
# Initialize resources on startup
db = await Database.connect()
try:
yield {"db": db}
finally:
# Clean up on shutdown
await db.disconnect()
# Pass lifespan to server
server = Server("example-server", lifespan=server_lifespan)
# Access lifespan context in handlers
@server.call_tool()
async def query_db(name: str, arguments: dict) -> list:
ctx = server.request_context
db = ctx.lifespan_context["db"]
return await db.query(arguments["query"])
La API de vida útil proporciona:
- Una forma de inicializar recursos cuando se inicia el servidor y limpiarlos cuando se detiene
- Acceso a recursos inicializados a través del contexto de solicitud en los controladores
- Transferencia de contexto de tipo seguro entre controladores de vida útil y de solicitudes
import mcp.server.stdio
import mcp.types as types
from mcp.server.lowlevel import NotificationOptions, Server
from mcp.server.models import InitializationOptions
# Create a server instance
server = Server("example-server")
@server.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:
return [
types.Prompt(
name="example-prompt",
description="An example prompt template",
arguments=[
types.PromptArgument(
name="arg1", description="Example argument", required=True
)
],
)
]
@server.get_prompt()
async def handle_get_prompt(
name: str, arguments: dict[str, str] | None
) -> types.GetPromptResult:
if name != "example-prompt":
raise ValueError(f"Unknown prompt: {name}")
return types.GetPromptResult(
description="Example prompt",
messages=[
types.PromptMessage(
role="user",
content=types.TextContent(type="text", text="Example prompt text"),
)
],
)
async def run():
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="example",
server_version="0.1.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
if __name__ == "__main__":
import asyncio
asyncio.run(run())
Escritura de clientes MCP
El SDK proporciona una interfaz de cliente de alto nivel para conectarse a servidores MCP:
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client
# Create server parameters for stdio connection
server_params = StdioServerParameters(
command="python", # Executable
args=["example_server.py"], # Optional command line arguments
env=None, # Optional environment variables
)
# Optional: create a sampling callback
async def handle_sampling_message(
message: types.CreateMessageRequestParams,
) -> types.CreateMessageResult:
return types.CreateMessageResult(
role="assistant",
content=types.TextContent(
type="text",
text="Hello, world! from model",
),
model="gpt-3.5-turbo",
stopReason="endTurn",
)
async def run():
async with stdio_client(server_params) as (read, write):
async with ClientSession(
read, write, sampling_callback=handle_sampling_message
) as session:
# Initialize the connection
await session.initialize()
# List available prompts
prompts = await session.list_prompts()
# Get a prompt
prompt = await session.get_prompt(
"example-prompt", arguments={"arg1": "value"}
)
# List available resources
resources = await session.list_resources()
# List available tools
tools = await session.list_tools()
# Read a resource
content, mime_type = await session.read_resource("file://some/path")
# Call a tool
result = await session.call_tool("tool-name", arguments={"arg1": "value"})
if __name__ == "__main__":
import asyncio
asyncio.run(run())
Primitivas MCP
El protocolo MCP define tres primitivos fundamentales que los servidores pueden implementar:
| Primitivo | Control | Descripción | Ejemplo de uso |
|---|
| Indicaciones | Controlado por el usuario | Plantillas interactivas invocadas por elección del usuario | Comandos de barra, opciones de menú |
| Recursos | Controlado por aplicación | Datos contextuales gestionados por la aplicación cliente | Contenido del archivo, respuestas de la API |
| Herramientas | Controlado por modelos | Funciones expuestas al LLM para tomar acciones | Llamadas API, actualizaciones de datos |
Capacidades del servidor
Los servidores MCP declaran capacidades durante la inicialización:
| Capacidad | Bandera de característica | Descripción |
|---|
prompts | listChanged | Gestión de plantillas de indicaciones |
resources | listChanged subscribe | Exposición y actualizaciones de recursos |
tools | listChanged | Descubrimiento y ejecución de herramientas |
logging | - | Configuración del registro del servidor |
completion | - | Sugerencias para completar argumentos |
Documentación
Contribuyendo
Nos apasiona apoyar a colaboradores de todos los niveles de experiencia y nos encantaría que participaras en el proyecto. Consulta la guía de colaboración para empezar.
Licencia
Este proyecto está licenciado bajo la licencia MIT: consulte el archivo de LICENCIA para obtener más detalles.