View a markdown version of this page

搭配代理程式架構使用 - Amazon OpenSearch Service
Strands 代理程式LangGraph

本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

搭配代理程式架構使用

您可以直接將 OpenSearch MCP 伺服器整合到 Python 代理程式架構中，讓您的自動代理程式以程式設計方式存取 OpenSearch，作為較大工作流程的一部分。

Strands 代理程式

Strands Agents 是 AWS原生代理程式 SDK，具有內建的 MCP 支援，並以 Amazon Bedrock 做為預設模型提供者。下列範例透過設定適當的環境變數，將 Strands 代理程式連線到 OpenSearch Service 網域或 OpenSearch Serverless 集合。連線至 OpenSearch Serverless 集合true時，請將 AWS_OPENSEARCH_SERVERLESS設定為 ；針對受管網域，請將其省略。

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)

字串使用 Amazon Bedrock 做為其預設模型提供者。請確定您已為區域中的 Claude 設定 AWS 登入資料並啟用模型存取。如需詳細資訊，請參閱 Strands Bedrock 提供者文件。

LangGraph

LangGraph 是用於建置具狀態代理程式的低階協同運作架構。下列範例使用 將 OpenSearch MCP 工具langchain-mcp-adapters載入 Amazon Bedrock 支援的 LangGraph ReAct 代理程式。如同 Strands，連線至 OpenSearch Serverless 集合true時，請將 AWS_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())

安裝必要的套件：

pip install langchain-aws langchain-mcp-adapters langgraph