Utilisation avec des frameworks d'agents

Vous pouvez intégrer le serveur OpenSearch MCP directement dans les frameworks d'agents Python, en donnant à vos agents autonomes un accès programmatique dans le OpenSearch cadre d'un flux de travail plus large.

Agents à mèches

Strands Agents est un SDK d'agent AWS natif avec support MCP intégré et Amazon Bedrock comme fournisseur de modèles par défaut. L'exemple suivant connecte un agent Strands à un domaine de OpenSearch service ou à une collection OpenSearch Serverless en définissant les variables d'environnement appropriées. Définissez cette AWS_OPENSEARCH_SERVERLESS valeur true lors de la connexion à une collection OpenSearch sans serveur ; omettez-la pour un domaine géré.

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 utilise Amazon Bedrock comme fournisseur de modèles par défaut. Assurez-vous que les AWS informations d'identification sont configurées et que l'accès aux modèles est activé pour Claude dans votre région. Pour plus de détails, consultez la documentation du fournisseur Strands Bedrock.

LangGraph

LangGraphest un framework d'orchestration de bas niveau permettant de créer des agents dynamiques. L'exemple suivant permet langchain-mcp-adapters de charger les outils OpenSearch MCP dans un LangGraph ReAct agent soutenu par Amazon Bedrock. Comme pour Strands, définissez ce paramètre AWS_OPENSEARCH_SERVERLESS sur true lors de la connexion à une collection OpenSearch Serverless.

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

Installez les packages requis :

pip install langchain-aws langchain-mcp-adapters langgraph