SDK de Python para MCP

Implementación en Python del Protocolo de Contexto de Modelo (MCP)

Tabla de contenido

Descripción general
Instalación
Inicio rápido
¿Qué es MCP?
Conceptos básicos
Ejecución de su servidor
Ejemplos
- Servidor Echo
- Explorador de SQLite
Uso avanzado
Documentación
Contribuyendo
Licencia

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

Recomendamos utilizar uv para administrar sus proyectos de Python:

uv add "mcp[cli]"

Alternativamente:

pip install mcp

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:

mcp install server.py

Alternativamente, puedes probarlo con el Inspector MCP:

mcp dev server.py

¿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.

from mcp.server.fastmcp import FastMCP

# Create a named server
mcp = FastMCP("My App")

# Specify dependencies for deployment and development
mcp = FastMCP("My App", dependencies=["pandas", "numpy"])

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.

@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:

@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:

@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[Message]:
    return [
        UserMessage("I'm seeing this error:"),
        UserMessage(error),
        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.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.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 = 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

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:

from mcp.server.fastmcp import FastMCP
import sqlite3

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:

from mcp.server.lowlevel import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types

# 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, Any] | 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
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
)

async def run():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) 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
            resource = 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.

FastMCP

Integrations