从 AgentCore 网关获取提示

以下 curl 请求显示了myTarget___code_review通过网关调用 ID mygateway-abcdefghij 获取提示的示例请求。

curl -X POST \
  https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -d '{
    "jsonrpc": "2.0",
    "id": "get-prompt-request",
    "method": "prompts/get",
    "params": {
      "name": "myTarget___code_review",
      "arguments": {
        "language": "python",
        "code": "print(\"hello\")"
      }
    }
}'

import requests
import json

def get_prompt(gateway_url, access_token, prompt_name, arguments):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {access_token}"
    }

    payload = {
        "jsonrpc": "2.0",
        "id": "get-prompt-request",
        "method": "prompts/get",
        "params": {
            "name": prompt_name,
            "arguments": arguments
        }
    }

    response = requests.post(gateway_url, headers=headers, json=payload)
    return response.json()

# Example usage
gateway_url = "https://${GatewayEndpoint}/mcp" # Replace with your actual gateway endpoint
access_token = "${AccessToken}" # Replace with your actual access token
result = get_prompt(
    gateway_url,
    access_token,
    "myTarget___code_review",  # Replace with {targetName}___{promptName}
    {"language": "python", "code": "print('hello')"}
)
print(json.dumps(result, indent=2))

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
import asyncio

async def execute_mcp(
    url,
    token,
    prompt_name,
    prompt_arguments,
    headers=None
):
    default_headers = {
        "Authorization": f"Bearer {token}"
    }
    headers = {**default_headers, **(headers or {})}

    async with streamablehttp_client(
       url=url,
       headers=headers,
    ) as (
        read_stream,
        write_stream,
        callA,
    ):
        async with ClientSession(read_stream, write_stream) as session:
            # 1. Perform initialization handshake
            print("Initializing MCP...")
            _init_response = await session.initialize()
            print(f"MCP Server Initialize successful! - {_init_response}")

            # 2. Get specific prompt
            print(f"Getting prompt: {prompt_name}")
            prompt_response = await session.get_prompt(
                name=prompt_name,
                arguments=prompt_arguments
            )
            print(f"Prompt response: {prompt_response}")
            return prompt_response

async def main():
    url = "https://${GatewayEndpoint}/mcp"
    token = "your_bearer_token_here"
    prompt_name = "myTarget___code_review"
    prompt_arguments = {
        "language": "python",
        "code": "print('hello')"
    }
    await execute_mcp(
        url=url,
        token=token,
        prompt_name=prompt_name,
        prompt_arguments=prompt_arguments
    )


if __name__ == "__main__":
    asyncio.run(main())

注意： LangGraph MCP 适配器提示支持可能有所不同。使用上面的 MCP 客户端方法实现最可靠的prompts/get实现。

import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async def get_prompt(url, token, prompt_name, arguments):
    headers = {"Authorization": f"Bearer {token}"}
    async with streamablehttp_client(url=url, headers=headers) as (
        read_stream, write_stream, callA
    ):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            response = await session.get_prompt(
                name=prompt_name,
                arguments=arguments
            )
            for message in response.messages:
                print(f"{message.role}: {message.content}")

asyncio.run(get_prompt(
    "https://${GatewayEndpoint}/mcp",
    "${AccessToken}",
    "myTarget___code_review",
    {"language": "python", "code": "print('hello')"}
))