Verwendung mit Agenten-Frameworks

Sie können den OpenSearch MCP-Server direkt in Python-Agent-Frameworks integrieren, sodass Ihre autonomen Agenten im OpenSearch Rahmen eines größeren Workflows programmatischen Zugriff darauf haben.

Strands, Agenten

Strands Agents ist ein AWS natives Agenten-SDK mit integrierter MCP-Unterstützung und Amazon Bedrock als Standardmodellanbieter. Das folgende Beispiel verbindet einen Strons-Agenten entweder mit einer OpenSearch Service-Domain oder einer OpenSearch Serverless-Sammlung, indem die entsprechenden Umgebungsvariablen festgelegt werden. trueWird AWS_OPENSEARCH_SERVERLESS auf gesetzt, wenn eine Verbindung zu einer OpenSearch serverlosen Sammlung hergestellt wird. Lassen Sie diese Einstellung bei einer verwalteten Domäne aus.

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 verwendet Amazon Bedrock als Standardmodellanbieter. Stellen Sie sicher, dass Sie die AWS Anmeldeinformationen konfiguriert und den Modellzugriff für Claude in Ihrer Region aktiviert haben. Einzelheiten finden Sie in der Anbieterdokumentation von Strands Bedrock.

LangGraph

LangGraphist ein Orchestrierungs-Framework auf niedriger Ebene zum Erstellen von Stateful-Agents. Das folgende Beispiel verwendetlangchain-mcp-adapters, um die OpenSearch MCP-Tools in einen LangGraph ReAct Agenten zu laden, der von Amazon Bedrock unterstützt wird. Stellen Sie wie bei Strands AWS_OPENSEARCH_SERVERLESS auf ein, true wenn Sie eine Verbindung zu einer OpenSearch serverlosen Sammlung herstellen.

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

Installieren Sie die erforderlichen Pakete:

pip install langchain-aws langchain-mcp-adapters langgraph