Model Context Protocol
& FastMCP

# STDIO (default) — local
mcp.run()

# HTTP — remoto
mcp.run(transport="http", host="0.0.0.0", port=8000)

# ASGI app (para Uvicorn/Gunicorn)
app = mcp.http_app()
# uvicorn app:app --host 0.0.0.0 --port 8000

from fastmcp import FastMCP

mcp = FastMCP("MeuServidor")

@mcp.tool
def buscar_cliente(cliente_id: str, incluir_historico: bool = False) -> dict:
    """Busca dados de um cliente pelo ID.

    Args:
        cliente_id: ID único do cliente
        incluir_historico: Se True, inclui histórico de pedidos
    """
    # sua lógica aqui
    return {"id": cliente_id, "nome": "João"}

from fastmcp import FastMCP, Context

@mcp.tool
async def processar_lote(itens: list[str], ctx: Context) -> str:
    """Processa uma lista de itens com progresso reportado."""
    for i, item in enumerate(itens):
        await ctx.report_progress(i, len(itens), f"Processando {item}")
        await ctx.info(f"Item {item} processado")
        # ... lógica ...
    return "Concluído"

from pydantic import BaseModel
from enum import Enum

class Status(Enum):
    ATIVO = "ativo"
    INATIVO = "inativo"

class FiltroCliente(BaseModel):
    status: Status
    limite: int = 10

@mcp.tool
def listar_clientes(filtro: FiltroCliente) -> list[dict]:
    """Lista clientes com filtro estruturado."""
    pass

# Exponha ou oculte tools por tags
@mcp.tool(tags={"public", "read-only"})
def tool_publica() -> str: ...

@mcp.tool(tags={"admin"})
def tool_admin() -> str: ...

# No servidor: só expõe tags "public"
mcp.enable(tags={"public"}, only=True)

@mcp.resource("config://app")
def get_config() -> dict:
    """Configuração atual da aplicação."""
    return {"versao": "1.0", "ambiente": "prod"}

@mcp.resource("usuario://{user_id}/perfil")
async def get_perfil(user_id: str) -> str:
    """Perfil de um usuário específico."""
    dados = await buscar_usuario(user_id)
    return json.dumps(dados)

# Cliente acessa: usuario://abc123/perfil

from fastmcp import Image

@mcp.resource("relatorio://{mes}/grafico")
async def get_grafico(mes: str) -> Image:
    img_bytes = await gerar_grafico(mes)
    return Image(data=img_bytes, format="png")

@mcp.prompt
def analisar_dados(pontos: list[float], contexto: str = "") -> str:
    """Template para análise de dados numéricos."""
    dados_fmt = ", ".join(str(p) for p in pontos)
    return f"""Analise estes dados: {dados_fmt}

Contexto adicional: {contexto}

Por favor identifique tendências, outliers e padrões relevantes."""

@mcp.tool
async def resumir_doc(uri: str, ctx: Context) -> str:
    # Lê resource
    conteudo = await ctx.read_resource(uri)

    # Pede ao LLM para resumir (sampling)
    resumo = await ctx.sample(
        f"Resuma em 3 bullets:\n{conteudo[0].text}",
        max_tokens=300
    )
    return resumo.text

from contextlib import asynccontextmanager
from fastmcp import FastMCP

@asynccontextmanager
async def lifespan(server):
    # Setup: roda quando servidor inicia
    db = await Database.connect("postgresql://...")
    redis = await Redis.create(...)

    yield {"db": db, "redis": redis}  # disponível via ctx.lifespan_context

    # Teardown: roda quando servidor para
    await db.disconnect()
    await redis.close()

mcp = FastMCP("MeuApp", lifespan=lifespan)

@mcp.tool
async def buscar(query: str, ctx: Context) -> list:
    db = ctx.lifespan_context["db"]
    return await db.fetch(query)

# Servidor de banco de dados
db_mcp = FastMCP("Database")

@db_mcp.tool
def query(sql: str) -> list: ...

# Servidor de arquivos
files_mcp = FastMCP("Files")

@files_mcp.tool
def ler_arquivo(path: str) -> str: ...

# Servidor principal que compõe os dois
main = FastMCP("Principal")
main.mount(db_mcp, prefix="db")
main.mount(files_mcp, prefix="files")

# Resultado: tools "db_query" e "files_ler_arquivo"

from fastmcp import FastMCP

# Gera servidor MCP a partir de qualquer OpenAPI spec
mcp = FastMCP.from_openapi(
    openapi_url="https://api.meuservico.com/openapi.json",
    client=httpx.AsyncClient(base_url="https://api.meuservico.com")
)
# Cada endpoint vira uma tool automaticamente!

import asyncio
from fastmcp import Client

# Em memória (mesmo processo — ideal para testes)
client = Client(meu_servidor)

# HTTP remoto
client = Client("https://mcp.meuservico.com/mcp")

# Script local (lança como subprocess via STDIO)
client = Client("./servidor.py", env={"API_KEY": "..."})

# Multi-servidor via config
client = Client({
    "mcpServers": {
        "db": {"url": "https://db.exemplo.com/mcp"},
        "files": {"command": "python", "args": ["./files.py"]}
    }
})

async with Client("https://mcp.exemplo.com/mcp") as client:
    # Listar
    tools     = await client.list_tools()
    resources = await client.list_resources()
    prompts   = await client.list_prompts()

    # Chamar tool
    resultado = await client.call_tool("buscar_cliente", {"cliente_id": "123"})
    print(resultado.data)

    # Ler resource
    conteudo = await client.read_resource("usuario://123/perfil")
    print(conteudo[0].text)

    # Renderizar prompt
    msgs = await client.get_prompt("analisar_dados", {"pontos": [1, 2, 3]})

    # Ping
    await client.ping()

from langchain_mcp_adapters.client import MultiServerMCPClient

async with MultiServerMCPClient({
    "agente_especialista": {
        "url": "https://agent.interno/mcp",
        "transport": "streamable_http",
        "headers": {"Authorization": f"Bearer {token}"}
    }
}) as client:
    tools = client.get_tools()  # LangChain BaseTool list
    agent = create_react_agent(llm, tools)

from fastmcp import Client

async def on_log(msg):
    print(f"[SERVER] {msg.level}: {msg.data}")

async def on_progress(progress, total, msg):
    pct = int(progress / total * 100) if total else "?"
    print(f"[PROGRESSO] {pct}% — {msg}")

client = Client(
    "https://mcp.exemplo.com/mcp",
    log_handler=on_log,
    progress_handler=on_progress,
    timeout=60.0
)

# No SERVIDOR: pede ao cliente para gerar texto
@mcp.tool
async def classificar(texto: str, ctx: Context) -> str:
    resposta = await ctx.sample(
        f"Classifique este texto em: positivo/negativo/neutro.\nTexto: {texto}",
        max_tokens=50
    )
    return resposta.text

# No CLIENTE: provê o handler de sampling
async def sampling_handler(messages, params, context):
    response = await llm.ainvoke(messages)
    return response.content

client = Client(mcp, sampling_handler=sampling_handler)

# server.py
from fastmcp import FastMCP

mcp = FastMCP("MeuServidor")

@mcp.custom_route("/health", methods=["GET"])
async def health(request):
    from starlette.responses import JSONResponse
    return JSONResponse({"status": "ok"})

if __name__ == "__main__":
    mcp.run(transport="http", host="0.0.0.0", port=8000)
# Endpoint MCP: http://host:8000/mcp

# app.py
app = mcp.http_app(path="/mcp")

# Dockerfile / ECS task
# uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4

from fastapi import FastAPI
from fastmcp import FastMCP

api = FastAPI()
mcp = FastMCP("MeuMCP")

# Suas rotas REST normais
@api.get("/usuarios")
async def usuarios(): ...

# MCP montado em /mcp
mcp_app = mcp.http_app()
api.mount("/mcp", mcp_app)
# uvicorn main:api

from fastmcp import FastMCP
from fastmcp.server.auth import BearerAuthProvider

# Valida JWTs emitidos pelo seu IdP via JWKS
auth = BearerAuthProvider(
    jwks_uri="https://seu-idp.com/.well-known/jwks.json",
    audience="https://sua-api.com",
    issuer="https://seu-idp.com/"
)

mcp = FastMCP("SecureServer", auth=auth)

# Acessar claims do token dentro das tools:
@mcp.tool
async def minha_tool(ctx: Context) -> str:
    token = ctx.get_access_token()
    user_id = token.claims.get("sub")
    return f"Olá, {user_id}"

from fastmcp import Client
from fastmcp.client.auth import BearerAuth

client = Client(
    "https://mcp.meuservico.com/mcp",
    auth=BearerAuth(token="eyJ...")
)

from fastmcp import FastMCP
from fastmcp.server.middleware import Middleware, MiddlewareContext

class LoggingMiddleware(Middleware):
    async def on_request(self, ctx: MiddlewareContext, call_next):
        print(f"→ {ctx.method} | client={ctx.client_id}")
        response = await call_next(ctx)
        print(f"← {ctx.method} ok")
        return response

mcp = FastMCP("App", middleware=[LoggingMiddleware()])

# CORS para HTTP (necessário apenas para clientes browser)
from starlette.middleware.cors import CORSMiddleware
from starlette.middleware import Middleware as StarletteMiddleware

app = mcp.http_app(middleware=[
    StarletteMiddleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"],
        allow_headers=["mcp-protocol-version", "mcp-session-id", "Authorization"],
        expose_headers=["mcp-session-id"])
])

# Habilitar no servidor
mcp = FastMCP("App", tasks=True)

@mcp.tool
async def processar_video(url: str, task=True) -> str:
    """Processa vídeo de forma assíncrona."""
    # Retorna imediatamente, roda em background
    resultado = await download_e_processar(url)
    return resultado

# Cliente: chama e observa progresso
async with client:
    task = await client.call_tool("processar_video", {"url": "..."})
    resultado = await client.wait_for_task(task.task_id)

import httpx
from fastmcp import FastMCP

# A partir de URL
mcp = FastMCP.from_openapi(
    openapi_url="https://api.exemplo.com/openapi.json",
    client=httpx.AsyncClient(
        base_url="https://api.exemplo.com",
        headers={"Authorization": "Bearer ..."}
    ),
    name="ExemploAPI"
)

# A partir de dict/arquivo local
with open("openapi.json") as f:
    spec = json.load(f)

mcp = FastMCP.from_openapi(openapi=spec, client=httpx.AsyncClient(...))

# Supervisor → Specialist (via Bearer JWT do IdP)
from langchain_mcp_adapters.client import MultiServerMCPClient

# Token obtido do IdP (Okta/Auth0) com escopo do agente
token = await get_agent_token(scope="specialist:financeiro")

client = MultiServerMCPClient({
    "financeiro": {
        "url": "https://financeiro-agent.interno/mcp",
        "transport": "streamable_http",
        "headers": {"Authorization": f"Bearer {token}"}
    }
})

# No specialist server: valida o JWT
auth = BearerAuthProvider(
    jwks_uri="https://seu-idp.com/.well-known/jwks.json",
    audience="specialist:financeiro"
)
mcp = FastMCP("Financeiro", auth=auth)

# Instalar servidor no Claude Desktop/Code
fastmcp install servidor.py

# Inspecionar tools/resources de um servidor
fastmcp inspect servidor.py

# Chamar tool direto do terminal
fastmcp call servidor.py buscar_cliente --cliente_id=123

# Gerar CLI tipado a partir de um servidor MCP
fastmcp generate-cli servidor.py -o cli_tool.py