Pular para conteúdo

Grupos de sessões

Tradução automática

Esta página foi traduzida automaticamente a partir da documentação em inglês, e a página em inglês é a versão de referência. Se algo parecer errado, Traduções explica como avisar.

Um Client se conecta a um servidor. Aplicações reais frequentemente querem vários (um servidor de busca, um servidor de banco de dados, uma API interna) e acabam fazendo malabarismo com uma conexão e uma lista de ferramentas (tools) para cada um.

ClientSessionGroup é um único objeto que mantém várias conexões e reúne tudo o que elas expõem em uma única visão.

Dois servidores

Comece com dois servidores comuns. Eles não têm nada a ver um com o outro, então ambos naturalmente chamaram sua ferramenta de search:

library_server.py
from mcp.server import MCPServer

mcp = MCPServer("Library")


@mcp.tool()
def search(query: str) -> str:
    """Search the library catalog."""
    return f"3 books match {query!r}."


@mcp.resource("library://hours")
def hours() -> str:
    """When the library is open."""
    return "Mon-Fri 09:00-17:00"
web_server.py
from mcp.server import MCPServer

mcp = MCPServer("Web")


@mcp.tool()
def search(query: str) -> str:
    """Search the web."""
    return f"12 pages match {query!r}."

Um grupo

Crie um ClientSessionGroup e chame connect_to_server uma vez por servidor:

client.py
import asyncio

from mcp import ClientSessionGroup, StdioServerParameters


async def main() -> None:
    library = StdioServerParameters(command="uv", args=["run", "mcp", "run", "library_server.py"])
    web = StdioServerParameters(command="uv", args=["run", "mcp", "run", "web_server.py"])

    async with ClientSessionGroup() as group:
        await group.connect_to_server(library)
        await group.connect_to_server(web)

        result = await group.call_tool("search", {"query": "model context protocol"})
        print(result.structured_content)


if __name__ == "__main__":
    asyncio.run(main())
  • connect_to_server recebe parâmetros de transporte, não um objeto de servidor: StdioServerParameters (de mcp) para iniciar um subprocesso, ou StreamableHttpParameters / SseServerParameters (de mcp.client.session_group) para um servidor que já está escutando em uma URL.
  • group.tools é um dict[str, Tool] com as ferramentas de todos os servidores conectados. group.resources e group.prompts têm o mesmo formato.
  • group.call_tool(name, arguments) procura o nome, encontra a sessão dona dele e encaminha a chamada. Você nunca diz qual servidor.

Check

Coloque client.py ao lado dos dois servidores e execute. O segundo connect_to_server recusa:

mcp.shared.exceptions.MCPError: {'search'} already exist in group tools.

Isso é um MCPError, lançado antes que qualquer coisa do segundo servidor seja registrada. Um nome precisa ser único no grupo inteiro, e dois servidores que você não controla vão colidir mais cedo ou mais tarde.

component_name_hook

Você resolve isso no grupo, não nos servidores. Passe uma função de (name, server_info) e o grupo a executa em cada nome que registra:

client.py
import asyncio

from mcp import ClientSessionGroup, StdioServerParameters
from mcp.types import Implementation


def by_server(name: str, server_info: Implementation) -> str:
    return f"{server_info.name}.{name}"


async def main() -> None:
    library = StdioServerParameters(command="uv", args=["run", "mcp", "run", "library_server.py"])
    web = StdioServerParameters(command="uv", args=["run", "mcp", "run", "web_server.py"])

    async with ClientSessionGroup(component_name_hook=by_server) as group:
        await group.connect_to_server(library)
        await group.connect_to_server(web)

        print(sorted(group.tools))
        result = await group.call_tool("Web.search", {"query": "model context protocol"})
        print(result.structured_content)


if __name__ == "__main__":
    asyncio.run(main())

Execute de novo. print(sorted(group.tools)) agora mostra as duas:

['Library.search', 'Web.search']
  • A chave é sua. by_server a montou a partir de server_info.name, o nome com que cada MCPServer(...) foi construído.
  • O Tool dentro fica intacto: group.tools["Web.search"].name ainda é "search", e esse é o nome que call_tool coloca na rede. O prefixo nunca sai do seu processo.
  • Não são só ferramentas. O recurso hours da biblioteca é registrado como Library.hours.

Tip

O hook é executado em cada nome de cada servidor, não só nos conflitos: não existe um modo de prefixar apenas em caso de colisão. Escolha um esquema e deixe que ele valha em todo lugar.

Adicionando e removendo servidores

connect_to_server retorna a ClientSession que abriu. Guarde-a se algum dia quiser tirar aquele servidor: await group.disconnect_from_server(session) remove do grupo as ferramentas, recursos e prompts dele.

Se você já tem em mãos uma ClientSession conectada (Client.session é uma), entregue-a a await group.connect_with_session(server_info, session) em vez de abrir um novo transporte. Ela é agregada da mesma forma. O grupo nunca fecha uma sessão que não abriu. server_info nomeia o servidor para os prefixos dos componentes; em uma conexão da era 2026, client.server_info pode ser None (a identidade é opcional), então nesse caso passe sua própria Implementation(name=..., version=...).

O handshake clássico

ClientSessionGroup é construído sobre ClientSession, não sobre Client. Cada connect_to_server executa o handshake clássico initialize. Ele nunca envia a sondagem server/discover descrita em Versões do protocolo. Todo servidor MCP entende esse handshake, então isso não custa compatibilidade com nada; significa apenas que um grupo segue o caminho mais antigo e mais lento até um servidor que poderia fazer melhor.

Recapitulando

  • ClientSessionGroup mantém várias conexões de servidor e reúne as ferramentas, recursos e prompts delas em um dict para cada tipo.
  • connect_to_server(params) por servidor. Ele recebe parâmetros de transporte, nunca o objeto de servidor ou a URL que um Client recebe.
  • group.call_tool(name, arguments) roteia para o servidor dono por você.
  • Os nomes precisam ser únicos no grupo inteiro; dois servidores com uma ferramenta search não conseguem coexistir por conta própria.
  • component_name_hook= reescreve cada nome registrado. A chave do dict muda, o nome na rede não.
  • connect_with_session adiciona uma sessão que você já tem; disconnect_from_server remove uma.

O handshake que um grupo fala (e o mais rápido que um Client prefere) é o assunto de Versões do protocolo.