Integrar MongoDB con el SDK del agente Claude

Overview

El paquete claude-agent-sdk-mongodb-session-store proporciona un adaptador de almacenamiento de sesiones con respaldo de MongoDB para las conversaciones del SDK de Claude Agent en su aplicación. Una sesión es un registro de la conversación y las acciones que realiza un agente, almacenado como entradas individuales. Por defecto, el SDK de Claude Agent almacena la transcripción en el host que invoca al agente. El adaptador de MongoDB también guarda esas entradas en MongoDB, por lo que la transcripción se conserva tanto en el host como en MongoDB. Posteriormente, puede reanudar la sesión desde otro host.

Puede utilizar el adaptador para realizar las siguientes tareas:

• Reanudar una sesión desde un proceso o host diferente.

• Mantén el historial de sesiones en la misma implementación que los datos de tu aplicación.

• Consulta y audita el historial de sesiones con herramientas de MongoDB

Instalación y configuración

Para utilizar el adaptador, debe tener disponibles los siguientes componentes:

• Python 3.10 o posterior

• Una implementación de MongoDB que ejecuta MongoDB 7.0 o posterior.

• Una clave API de Anthropic

Para obtener y configurar su clave API de Anthropic, consulte la Guía de inicio rápido del SDK de Claude Agent. El almacén de sesiones no utiliza ni tiene acceso a esta clave.

En tu terminal, ejecuta el siguiente comando para instalar el adaptador y sus dependencias:

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

Este comando instala el adaptador y sus dependencias claude-agent-sdk y pymongo.

Configure su cadena de conexión en una variable de entorno, como se muestra en el siguiente ejemplo. El adaptador utiliza el cliente que usted crea, por lo que sus credenciales permanecen en la configuración de su cliente:

export MONGODB_URI="<connection-string>"

Persistir una sesión

Por defecto, el SDK del agente Claude guarda cada transcripción de sesión únicamente en el host donde se ejecuta el agente. El siguiente ejemplo crea un almacén de sesiones de MongoDB, inicializa sus índices y ejecuta una consulta. El adaptador guarda la transcripción de la sesión en MongoDB mientras el agente se ejecuta. El ejemplo imprime el resultado y el ID de sesión asignado por el SDK.

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())

Reanudar una sesión

Para reanudar una sesión, cree un objeto ClaudeAgentOptions y pásele el objeto de almacenamiento y el ID de sesión de la sesión que desea reanudar. El SDK devuelve el ID de sesión en el campo session_id del mensaje de resultado de la primera ejecución.

Puedes usar el siguiente fragmento de código dentro de una función asíncrona para reanudar una sesión. Reemplaza <session-id> con tu ID de sesión:

# 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)

Consideraciones operativas y de retención

El adaptador almacena cada entrada de transcripción como un documento y nunca elimina las sesiones por sí solo. Para eliminar las sesiones que ya no están activas, llame a delete_inactive() según una programación, por ejemplo, desde una tarea cron o un activador programado de 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.")

No elimine entradas individuales con un índice de tiempo de vida o un filtro delete_many() en la hora de entrada. Eliminar las entradas más antiguas de una sesión que aún está en uso deja una transcripción que su aplicación no puede reanudar.

Recursos adicionales

Para obtener más información sobre el adaptador de almacenamiento de sesiones de MongoDB y el SDK de Claude Agent, consulte los siguientes recursos:

• Documentación sobre el almacenamiento de sesiones del SDK de Claude Agent

• Repositorio del adaptador de almacenamiento de sesiones de MongoDB

Visita el repositorio para mantenerte al día sobre el adaptador y considera marcarlo como favorito.