MCP (Model Context Protocol)

Conecte servidores MCP às chamadas via mcp_servers=[...]. Todos os providers suportam MCP, mas de formas diferentes — e a jangada usa o jeito nativo de cada SDK.

Dois modelos (importante)

• Remoto (URL) — MCPServer(url=...): o provider conecta no servidor MCP e executa as tools (server-side). Você não roda nada. → Anthropic, OpenAI, Groq.
• Client-side (sessão) — você passa uma ClientSession do pacote mcp: o SDK chama as tools localmente (automatic function calling). → Gemini.

Remoto (Anthropic / OpenAI / Groq)

from jangada_ai import LLM, MCPServer

llm = LLM("anthropic", "claude-opus-4-8")   # ou ("openai", "gpt-4o"), ("groq", ...)
comp = llm.complete(
    "Liste as issues abertas do repositório.",
    mcp_servers=[MCPServer(
        url="https://mcp.exemplo.com/sse",
        name="github",
        authorization_token="TOKEN",     # opcional (OAuth/Bearer)
        allowed_tools=["list_issues"],   # opcional (restringe as tools)
    )],
)
print(comp.text)   # o provider já executou as tools do MCP

Client-side (Gemini, async)

No Gemini, MCPServer(url=...) também funciona — mas só em acomplete: a jangada abre um MCPClient próprio por baixo e entrega ao SDK como sessão client-side (é o único jeito que o Gemini suporta). Precisa do extra [mcp].

from jangada_ai import LLM, MCPServer

llm = LLM("gemini", "gemini-2.5-flash")
comp = await llm.acomplete(
    "Liste as issues abertas.",
    mcp_servers=[MCPServer(url="https://mcp.exemplo.com/mcp/", name="github",
                           authorization_token="TOKEN")],
)

No complete() (sync) isso continua levantando UnsupportedError — o Gemini não tem como abrir uma sessão assíncrona fora de acomplete. allowed_tools/ require_approval do MCPServer também levantam UnsupportedError nesse caminho (em vez de serem ignorados em silêncio) — o SDK do Gemini lista e executa as tools sozinho, sem hook de filtro/aprovação; para restringir tools no Gemini, use o MCPClient/run_agent próprio da jangada (abaixo), que tem allowed_tools=.

A sessão MCP é assíncrona (montada manualmente), então use acomplete:

from mcp import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters
from jangada_ai import LLM

llm = LLM("gemini", "gemini-2.5-flash")
params = StdioServerParameters(command="npx", args=["-y", "@exemplo/mcp"])

async with stdio_client(params) as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()
        comp = await llm.acomplete("Use a ferramenta X", mcp_servers=[session])
        print(comp.text)

No Gemini, complete() (sync) com sessão levanta UnsupportedError pedindo acomplete(). E passar uma sessão no Anthropic/OpenAI/Groq levanta — eles são remotos por URL.

Cliente MCP próprio + agente (portável, qualquer provider)

A jangada traz seu próprio cliente MCP (MCPClient) e um loop de agente (run_agent) que conecta no servidor, lista as tools, e roda o ciclo (modelo pede → executa → reenvia) sozinho — em qualquer provider (usa tools=, suportado nos 4), independente do MCP nativo de cada SDK.

pip install "jangada-ai[mcp]"   # cliente MCP (pacote oficial `mcp`)

from jangada_ai import LLM
from jangada_ai.mcp import MCPClient, run_agent

llm = LLM("openai", "gpt-4o-mini")   # ou anthropic/groq/gemini

async with MCPClient("https://meu-mcp/mcp/") as mcp:      # ou command=/args= (stdio)
    ans = await run_agent(llm, "Role uns dados", client=mcp)
    print(ans.text)

Por baixo: await mcp.list_tools() vira tools=[...], e cada tool_call do modelo é executado com await mcp.call_tool(...) e reenviado via Message.tool_results(...) — o mesmo tool calling de sempre, no automático. Quer controle total? Use MCPClient + tools= na mão.

• isError do resultado da tool vira is_error=True no tool_result — o modelo sabe que a tool falhou (diferente de uma exceção de rede/protocolo, que já virava erro antes).
• list_tools/list_resources/list_prompts seguem o nextCursor sozinhos até esgotar as páginas — servidores com muitas tools não ficam incompletos. Param se um servidor com bug repetir o mesmo cursor, e têm um limite defensivo de 10.000 páginas para cursores que nunca se repetem mas também nunca esgotam — batendo nesse limite sem esgotar o cursor emite um UserWarning (a lista pode estar incompleta; nunca trunca em silêncio).
• Limite de iterações: se run_agent bater em max_iterations com tool_calls ainda pendentes, ele emite um UserWarning (Completion devolvido não é necessariamente a resposta final). No Agent.run/arun (abaixo), o mesmo caso também marca AgentResult.stopped_by_limit = True.
• Erro de conexão (MCPClient.__aenter__, ex.: token inválido, servidor fora do ar) vira um APIConnectionError da própria lib com a causa real (ex.: HTTP 403/HTTP 400), em vez do erro cru do transporte (ExceptionGroup/CancelledError do anyio) — mesmo quando a causa real só aparece ao fechar a conexão, não na abertura.

Transporte, autenticação e timeout

async with MCPClient(
    "https://meu-mcp/sse",     # URL terminando em /sse -> detecta SSE sozinho
    # transport="sse",         # ou force explicitamente ("sse" | "streamable-http")
    auth_token="TOKEN",        # vira header Authorization: Bearer TOKEN
    timeout=30,                # segundos, repassado ao transporte HTTP
) as mcp:
    ...

Conexão de vida longa (connect/aclose/reconnect, keep_alive, ping)

Fora do async with — útil para abrir no startup da aplicação e fechar no shutdown:

mcp = MCPClient("https://meu-mcp/mcp/", auth_token="TOKEN")
await mcp.connect()          # idempotente: chamar de novo já conectado é no-op
...
await mcp.ping()             # health check (session/ping)
...
await mcp.reconnect()        # fecha e reabre (ex.: percebeu a conexão caída)
...
await mcp.aclose()           # no shutdown

Com keep_alive=True, o MCPClient conecta sozinho na 1ª chamada (sem precisar de connect()/async with) e reconecta uma vez, sozinho, se uma chamada falhar com algo que parece conexão caída — não repete em erro de LÓGICA da tool (ex.: argumento inválido):

mcp = MCPClient("https://meu-mcp/mcp/", auth_token="TOKEN", keep_alive=True)
tools = await mcp.list_tools()   # conecta sozinho aqui

Seguro para uso concorrente: se várias chamadas percebem a conexão caída ao mesmo tempo, só a primeira reconecta de verdade — as outras esperam e reaproveitam a sessão nova, em vez de disparar reconexões por cima umas das outras. Se a reconexão em si falhar (servidor fora do ar), as chamadas que esperavam reusam o MESMO erro por um cooldown curto, em vez de cada uma esperar seu próprio timeout de conexão do zero.

Primitivos completos do MCP (no MCPClient)

Além de tools, o MCPClient cobre o resto do protocolo:

async with MCPClient("https://meu-mcp/mcp/") as mcp:
    # Resources — dados/contexto que o server expõe
    recursos = await mcp.list_resources()
    texto    = await mcp.resource_text("file:///guia.md")

    # Prompts — templates reutilizáveis do server -> vira list[Message]
    msgs = await mcp.prompt_messages("revisar", {"texto": "..."})
    resp = await llm.acomplete(None, history=msgs)

E os recursos de cliente (o server chama o cliente de volta), configurados no construtor:

from jangada_ai import LLM

async with MCPClient(
    command="python", args=["server.py"],
    roots=["./workspace"],                       # escopo de filesystem (file://)
    sampling_llm=LLM("openai", "gpt-4o-mini"),   # o server pede geração ao SEU LLM
    elicitation_callback=meu_handler,            # o server pede input ao usuário
    logging_callback=meu_logger,                 # logs do server
) as mcp:
    ...

• Roots: o server pergunta quais diretórios pode usar; o cliente responde a lista.
• Sampling: o server pede uma geração de LLM (sampling/createMessage) e a jangada executa com o seu LLM — o server fica model-independent e você controla custo/permissões.
• Elicitation / Logging: você passa um callback (async) que o SDK chama.

run_agent: histórico, callbacks, allowlist e retorno enriquecido

from jangada_ai.message import Message

def veta_apagar(call):
    return call.name != "apagar_arquivo"   # False = veta a chamada

async with MCPClient("https://meu-mcp/mcp/") as mcp:
    ans = await run_agent(
        llm, "Liste e depois apague os temporários", client=mcp,
        history=[Message("user", "oi"), Message("assistant", "olá!")],  # turnos anteriores
        allowed_tools=["listar_arquivos", "apagar_arquivo"],  # restringe as tools visíveis
        on_tool_call=veta_apagar,        # (sync ou async) False = veta a chamada
        on_tool_result=lambda c, r: print(c.name, r.is_error),
    )
    print(ans.text, ans.iterations, ans.stopped_by_limit)
    print(ans.tool_trace)          # [{"call", "result", "is_error"}, ...] de TODAS as rodadas
    print(ans.usage_total, ans.cost_total, ans.cost_complete)   # agregados do loop inteiro

• history= injeta turnos anteriores antes da tarefa; prompt=None continua só do histórico (sem adicionar mensagem de usuário vazia).
• allowed_tools= filtra pelo nome antes do modelo ver as tools (também disponível em mcp_tools(client, allowed_tools=[...]) direto).
• tools= pula o list_tools() (e o round-trip) quando você já listou antes — liste uma vez e reuse entre chamadas; allowed_tools= continua sendo aplicado por cima de tools= (filtra a lista já pronta).
• on_tool_call(call)/on_tool_result(call, result) (sync ou async) correm a cada tool call; on_tool_call devolvendo False veta a chamada.
• O Completion devolvido ganha atributos extras (usage/cost continuam sendo só os da ÚLTIMA chamada ao LLM):

Em Agent/Squad, o mesmo aparece como mcp_allowed_tools=, on_tool_call=/on_tool_result= no construtor, e AgentResult.tool_trace.

Ser um servidor MCP (expor suas tools/Agent)

O jangada também pode ser um servidor MCP — expor suas funções/Agent para clientes (Claude Desktop, Cursor, outro agente). Construído sobre o Server low-level do protocolo (não FastMCP); o schema sai do tools.py.

from jangada_ai import serve_mcp

def somar(a: float, b: float) -> float:
    "Soma dois números."
    return a + b

serve_mcp("minha-calc", tools=[somar])          # stdio (Claude Desktop/Cursor)

Expor um Agent (vira a tool ask):

serve_mcp("suporte", agent=Agent(LLM("openai", "gpt-4o-mini"), role="suporte"))

Por HTTP (streamable-http) ou montando no seu ASGI:

serve_mcp("minha-calc", tools=[somar], transport="streamable-http", port=8000)
# ou: app = build_mcp_app("minha-calc", tools=[somar], path="/mcp")  # Starlette

Extra jangada-ai[mcp] (HTTP também precisa de starlette + uvicorn). serve_mcp/build_mcp_app = servidor; MCPClient = cliente.

Exemplo pronto: jangada-docs-mcp serve toda a doc do jangada ao seu editor (Claude Code/Desktop/Cursor) — sem clonar: uvx --from git+https://github.com/nerigleston/jangada-docs-mcp jangada-docs-mcp.

O que mudou na 1.9.0

• Compatível com mcp 1.x e 2.x. O extra [mcp] agora aceita mcp>=1.0,<3; cliente, servidor, SSE, streamable-HTTP e stdio foram validados de ponta a ponta nas duas versões.
• Retry só no que é seguro repetir. Com keep_alive=True, a reconexão automática repete sozinha apenas operações sem efeito colateral (list_*, read_resource, get_prompt, ping). call_tool só é repetido quando a requisição comprovadamente não saiu (falha de conexão); em timeout de leitura ele reconecta para as próximas chamadas mas não repete a atual (evita criar um pedido em dobro). Para repetir mesmo assim: MCPClient(..., retry_tools=True). Erros 401/403 (e demais 4xx, exceto 408/429) não disparam reconexão.
• timeout= vale para a sessão inteira, inclusive no stdio — uma tool travada no servidor não pendura mais o cliente para sempre.
• Allowlist aplicada na execução. run_agent(..., allowed_tools=[...]) (e o Agent com mcp_allowed_tools) só executa tools que foram de fato oferecidas ao modelo; qualquer outro nome volta como tool_result de erro, sem executar.
• SSE detectado pelo caminho da URL (/sse?token=... funciona).
• Sampling com aprovação. MCPClient(..., sampling_llm=llm, on_sampling_request=fn): fn(request) (sync ou async) devolvendo False recusa o pedido do servidor. O stopReason reflete o motivo real (maxTokens, endTurn, toolUse) e prompt_messages preserva imagens.
• Segredos fora do repr. Token e headers não aparecem mais em repr(MCPClient) nem em repr(MCPServer).

Servidor: autenticação e proteção contra DNS rebinding

from jangada_ai import serve_mcp

serve_mcp(
    "minhas-tools", tools=[consultar_pedido],
    transport="streamable-http", host="0.0.0.0", port=8000,
    auth_token="segredo",                      # exige Authorization: Bearer segredo (401 sem ele)
    allowed_hosts=["mcp.minhaempresa.com"],     # proteção contra DNS rebinding
    allowed_origins=["https://app.minhaempresa.com"],
)

build_mcp_app(...) aceita os mesmos auth_token/allowed_hosts/ allowed_origins/security_settings. Tools síncronas (e agente síncrono) rodam numa thread, sem bloquear as outras sessões. Se o agent= exposto tiver memory, a lib avisa: a memória é compartilhada entre todos os clientes.

Exemplo

• examples/mcp_example.py — cliente MCP.
• examples/mcp_server_example.py — servidor MCP (stdio).