水曜日, 6月 4, 2025
- Advertisment -
ホームニューステックニュースひとつのLambdaの中でMCPサーバーとクライアントをStdio通信させる #AWS - Qiita

ひとつのLambdaの中でMCPサーバーとクライアントをStdio通信させる #AWS – Qiita



ひとつのLambdaの中でMCPサーバーとクライアントをStdio通信させる #AWS - Qiita

Strands AgentsはMCPに対応してます。

以下のような最低限のコードでMCPツールを使用できます。

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

# Connect to an MCP server using stdio transport
# Note: uvx command syntax differs by platform

# For macOS/Linux:
stdio_mcp_client = MCPClient(lambda: stdio_client(
    StdioServerParameters(
        command="uvx", 
        args=["awslabs.aws-documentation-mcp-server@latest"]
    )
))

# Create an agent with MCP tools
with stdio_mcp_client:
    # Get the tools from the MCP server
    tools = stdio_mcp_client.list_tools_sync()

    # Create an agent with these tools
    agent = Agent(tools=tools)

いいですね。病みつきですね。

Lambda化する

AWS好きなみなさんが次に考えることは、そう、「これ、Lambdaで動かしたい」ですよね。

で、SAMを使ってこのままLambdaにしてみます。

app.py

import json

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


def lambda_handler(event, context):
    prompt = event["prompt"]

    stdio_mcp_client = MCPClient(
        lambda: stdio_client(
            StdioServerParameters(
                command="uvx", 
                args=["awslabs.aws-documentation-mcp-server@latest"]
            )
        )
    )

    with stdio_mcp_client:
        # Get the tools from the MCP server
        tools = stdio_mcp_client.list_tools_sync()

        # Create an agent with these tools
        agent = Agent(tools=tools)

        result = agent(prompt=prompt)

    return {
        "statusCode": 200,
        "body": json.dumps(
            {"message": result.message["content"][0]["text"]}, ensure_ascii=False
        ),
    }

requirements.txt

strands-agents
awslabs.aws-documentation-mcp-server

event["prompt"]を入力にしますので、イベントJSONに以下の値を入れてテスト実行してみます。

{
  "prompt": "S3のバケットポリシーについてドキュメントを検索して"
}

すると、残念ながらエラーになります。

せっかくなので「Diagnose with Amazon Q」ボタンを押してみましょう。

Check if the ‘uvx’ executable is present in the Lambda function’s execution environment

流石Q Developerです。uvxコマンドが無いからエラーになったんですね。

じゃあ、どうすんの?

uvxコマンドをインストールするのもいいですが、そもそもこの「uvx」って何者なんでしょうか??

ドキュメントはここにありました。

Many Python packages provide applications that can be used as tools. uv has specialized support for easily invoking and installing tools.

日本語に訳すと、

多くの Python パッケージは、ツールとして使用できるアプリケーションを提供します。uv は、ツールを簡単に呼び出してインストールするための特別なサポートを備えています。

だとのこと。で、ツールってなんなの??

これについてはPythonのドキュメントに記載がありました。

上で例に出したAWS Documentation MCP Serverの場合、pyproject.tomlには以下のように記述があります。

pyproject.toml

[project.scripts]
"awslabs.aws-documentation-mcp-server" = "awslabs.aws_documentation_mcp_server.server:main"

要するに、uvx awslabs.aws-documentation-mcp-serverを実行すると

「指定したライブラリーをいい感じにダウンロードして、project.scriptsに記述のあるスクリプトawslabs.aws_documentation_mcp_server.server:mainを実行するよ

ということだと思われます。
ちなみに、awslabs.aws_documentation_mcp_serverがパッケージでserverがモジュール、mainは関数を指しています。

ここまでわかったら、uvxコマンドを使用しなくても、pythonコマンドで実行できます。

uvx awslabs.aws-documentation-mcp-server

python -m awslabs.aws_documentation_mcp_server.server

と同義です。

最後の:mainの指定がないですが、-mオプションでモジュール名を指定する場合、関数名を指定することはできません。ただ、server.pyに以下の記述があるため、結局main関数が呼ばれる動作になります。

if __name__ == '__main__':
    main()

ということで動作するコード

python -m awslabs.aws_documentation_mcp_server.serverをMCPのSDKに合わせた形で指定します。

import json

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


def lambda_handler(event, context):
    prompt = event["prompt"]

    stdio_mcp_client = MCPClient(
        lambda: stdio_client(
            StdioServerParameters(
                command="python",
                args=["-m", "awslabs.aws_documentation_mcp_server.server"],
            )
        )
    )

    with stdio_mcp_client:
        # Get the tools from the MCP server
        tools = stdio_mcp_client.list_tools_sync()

        # Create an agent with these tools
        agent = Agent(tools=tools)

        result = agent(prompt=prompt)

    return {
        "statusCode": 200,
        "body": json.dumps(
            {"message": result.message["content"][0]["text"]}, ensure_ascii=False
        ),
    }

わーい😀

4分かかったけど

ソースおいときますね





Source link

Views: 2

RELATED ARTICLES

返事を書く

あなたのコメントを入力してください。
ここにあなたの名前を入力してください

- Advertisment -