Integre o MongoDB com o Class Agent SDK

Visão geral

O pacote claude-agent-sdk-mongodb-session-store fornece um adaptador de armazenamento de sessão apoiado MongoDB para conversas do SDK do. Uma sessão é um registro da conversa e das ações que um agente realiza, armazenado como entradas individuais. Por padrão, o Class SDK do agente armazena a transcrição no host que invoca o agente. O adaptador MongoDB também salva essas entradas no MongoDB, para que a transcrição seja mantida no host e no MongoDB. Você pode então retomar a sessão de outro host.

Você pode usar o adaptador para executar as seguintes tarefas:

• Retomar uma sessão de um processo ou host diferente

• Mantenha o histórico de sessões no mesmo sistema que os dados do aplicação

• Query e histórico de sessões de auditar com ferramentas MongoDB

Instalação e configurar

Para usar o adaptador, você deve ter os seguintes componentes disponíveis:

• Python 3.10 ou posterior

• Uma implantação do MongoDB que executa o MongoDB 7.0 ou posterior

• Uma chave de API antrófica

Para saber como obter e definir sua chave de API antrófica, consulte o Início rápido do SDK do agente. O armazenamento de sessão não usa nem vê esta chave.

No seu terminal, execute o seguinte comando para instalar o adaptador e suas dependências:

uv add claude-agent-sdk-mongodb-session-store

Este comando instala o adaptador e suas dependências claude-agent-sdk e pymongo.

Defina sua string de conexão em uma variável de ambiente, como mostrado no exemplo a seguir. O adaptador usa o cliente que você cria, portanto, suas credenciais permanecem na configuração do cliente :

export MONGODB_URI="<connection-string>"

Persistir uma sessão

Por padrão, o Class SDK do agente mantém cada transcrição de sessão apenas no host onde o agente é executado. O exemplo a seguir cria um armazenamento de sessão MongoDB , inicializa seus índices e executa uma query. O adaptador mantém a transcrição da sessão para o MongoDB à medida que o agente é executado. O exemplo imprime o resultado e o ID da sessão que o SDK atribui.

import asyncio
import os
from pymongo import AsyncMongoClient
from claude_agent_sdk import ClaudeAgentOptions, ResultMessage, query
from claude_agent_sdk_mongodb_session_store import MongoDBSessionStore
async def main() -> None:
    client = AsyncMongoClient(
        os.environ["MONGODB_URI"],
        readPreference="primary",
        w="majority",
        serverSelectionTimeoutMS=10_000,
        timeoutMS=20_000,
    )
    store = MongoDBSessionStore(client=client, db_name="claude")
    # Set up the database indexes that optimize the store's performance.
    # This operation is idempotent, so you can run it on every startup.
    await store.create_schema()
    async for message in query(
        prompt="Review this codebase and explain what it does.",
        options=ClaudeAgentOptions(session_store=store),
    ):
        if isinstance(message, ResultMessage) and message.subtype == "success":
            print(message.result)
            # The SDK assigns the session ID. Save it so that you can
            # resume the session later.
            print(f"Session ID: {message.session_id}")
    await client.close()
asyncio.run(main())

Retomar uma sessão

Para retomar uma sessão, crie um objeto ClaudeAgentOptions e passe o objeto de armazenamento e o ID de sessão da sessão que você deseja retomar. O SDK retorna a ID da sessão no campo session_id da mensagem de resultado da primeira execução.

Você pode usar o seguinte trecho dentro de uma função assíncrona para retomar uma sessão. Substitua <session-id> pelo ID da sua sessão:

# Pass the same store and the session ID from the earlier run.
async for message in query(
    prompt="Suggest how to improve this codebase.",
    options=ClaudeAgentOptions(
        session_store=store,
        resume="<session-id>",
    ),
):
    if isinstance(message, ResultMessage) and message.subtype == "success":
        print(message.result)

Retenção e considerações operacionais

O adaptador armazena cada entrada da transcrição como um documento e nunca exclui sessões por conta própria. Para remover sessões que não estão mais ativas, chame delete_inactive() em uma programação, como de uma tarefa do cron ou de um trigger agendado do Atlas :

from datetime import timedelta
# Delete every session with no new entries in the last 30 days.
deleted = await store.delete_inactive(timedelta(days=30))
print(f"Deleted {deleted} inactive sessions.")

Não exclua entradas individuais com um índice de duração ou um filtro delete_many() no tempo de entrada. A remoção das entradas mais antigas de uma sessão que ainda está em uso deixa uma transcrição que seu aplicação não pode retomar.

Recursos adicionais

Para saber mais sobre o adaptador de armazenamento de sessão MongoDB e o SDK do agente Class, consulte os seguintes recursos:

• Documentação de armazenamento de sessão do Class SDK

• Repositório do adaptador de armazenamento de sessão do MongoDB

Visite o repositório para manter-se atualizado sobre o adaptador e considere definir o repositório como principal.