Úselo con marcos de agentes

Puede integrar el servidor OpenSearch MCP directamente en los marcos de agentes de Python, lo que permite a sus agentes autónomos acceder mediante programación OpenSearch como parte de un flujo de trabajo más amplio.

Strands Agents

Strands Agents es un SDK AWS de agente nativo con soporte MCP integrado y Amazon Bedrock como proveedor de modelos predeterminado. El siguiente ejemplo conecta un agente de Strands a un dominio de OpenSearch servicio o a una colección OpenSearch Serverless mediante la configuración de las variables de entorno adecuadas. AWS_OPENSEARCH_SERVERLESStrueEstablézcalo en Al conectarse a una colección OpenSearch sin servidor; omítalo en el caso de un dominio administrado.

import os
from strands import Agent
from strands.tools.mcp import MCPClient
from mcp import stdio_client, StdioServerParameters

# For a managed domain:
#   OPENSEARCH_URL = https://<domain-endpoint>.<region>.es.amazonaws.com
#
# For an OpenSearch Serverless collection, also set:
#   AWS_OPENSEARCH_SERVERLESS = true
#   OPENSEARCH_URL = https://<collection-id>.<region>.aoss.amazonaws.com
opensearch_client = MCPClient(
    lambda: stdio_client(
        StdioServerParameters(
            command="uvx",
            args=["opensearch-mcp-server-py"],
            env={
                "OPENSEARCH_URL":            os.environ["OPENSEARCH_URL"],
                "AWS_REGION":                os.environ["AWS_REGION"],
                "AWS_IAM_ARN":               os.environ["AWS_IAM_ARN"],
                # Set to "true" for OpenSearch Serverless, omit for managed domains
                "AWS_OPENSEARCH_SERVERLESS": os.environ.get("AWS_OPENSEARCH_SERVERLESS", "false"),
            },
        )
    )
)

with opensearch_client:
    agent = Agent(tools=opensearch_client.list_tools_sync())
    response = agent("List all indexes and show the document count for each")
    print(response)

Strands utiliza Amazon Bedrock como proveedor de modelos predeterminado. Asegúrese de tener AWS las credenciales configuradas y de habilitar el acceso a modelos para Claude en su región. Para obtener más información, consulte la documentación del proveedor de Strands Bedrock.

LangGraph

LangGraphes un marco de orquestación de bajo nivel para crear agentes con estado. El siguiente ejemplo se utiliza langchain-mcp-adapters para cargar las herramientas de OpenSearch MCP en un LangGraph ReAct agente respaldado por Amazon Bedrock. Al igual que con Strands, se configura AWS_OPENSEARCH_SERVERLESS en true cuando se conecta a una colección OpenSearch sin servidor.

import asyncio
import os
from langchain_aws import ChatBedrock
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent

async def main():
    async with MultiServerMCPClient(
        {
            "opensearch": {
                "command": "uvx",
                "args": ["opensearch-mcp-server-py"],
                "env": {
                    # Managed domain:  https://<domain-endpoint>.<region>.es.amazonaws.com
                    # Serverless:      https://<collection-id>.<region>.aoss.amazonaws.com
                    "OPENSEARCH_URL":            os.environ["OPENSEARCH_URL"],
                    "AWS_REGION":                os.environ["AWS_REGION"],
                    "AWS_IAM_ARN":               os.environ["AWS_IAM_ARN"],
                    # Set to "true" for OpenSearch Serverless, omit for managed domains
                    "AWS_OPENSEARCH_SERVERLESS": os.environ.get("AWS_OPENSEARCH_SERVERLESS", "false"),
                },
                "transport": "stdio",
            }
        }
    ) as mcp_client:
        tools = mcp_client.get_tools()
        model = ChatBedrock(
            model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
            region_name=os.environ["AWS_REGION"],
        )
        agent = create_react_agent(model, tools)
        result = await agent.ainvoke(
            {"messages": [{"role": "user", "content": "Check cluster health and list all indexes"}]}
        )
        print(result["messages"][-1].content)

asyncio.run(main())

Instale los paquetes obligatorios:

pip install langchain-aws langchain-mcp-adapters langgraph