### Run MCP Server with Default Stdio Source: https://github.com/davidshtian/mcp-on-aws-bedrock/blob/hand/README.md Execute the MCP server using default stdio settings. This is the basic setup for running the server without specifying a transport type. ```bash uv run fetch_url_mcp_server.py ``` -------------------------------- ### Run MCP Server with Streamable HTTP Source: https://github.com/davidshtian/mcp-on-aws-bedrock/blob/hand/README.md Start the MCP server using the streamable-http transport on the default port. This enables enhanced HTTP communication. ```bash python fetch_url_mcp_server.py --transport streamable-http ``` -------------------------------- ### Run MCP Client with Stdio Source: https://github.com/davidshtian/mcp-on-aws-bedrock/blob/hand/README.md Execute the MCP client using stdio for communication. This pairs with the server running in default stdio mode. ```bash uv run client_stdio.py ``` -------------------------------- ### FastMCP Server for URL Fetching Tool Source: https://context7.com/davidshtian/mcp-on-aws-bedrock/llms.txt Implements a FastMCP server that exposes a `fetch_url` tool using `httpx`. Supports stdio and Streamable HTTP transports, configurable via command-line arguments. ```python # fetch_url_mcp_server.py — run directly, do not import # Start with stdio transport (default): # uv run fetch_url_mcp_server.py # Start with Streamable HTTP on port 8000 (default): # python fetch_url_mcp_server.py --transport streamable-http # Start with Streamable HTTP on a custom port: # python fetch_url_mcp_server.py --transport streamable-http --port 8080 import httpx import argparse from pydantic import Field from mcp.server.fastmcp import FastMCP def create_fetch_url_tool(mcp): @mcp.tool() async def fetch_url( url: str = Field(description="URL to fetch"), ) -> str: """Fetches a website and returns its content""" async with httpx.AsyncClient() as client: response = await client.get(url) return response.text return fetch_url if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--transport", type=str) parser.add_argument("--port", type=int, default=8000) args = parser.parse_args() if args.transport == "streamable-http": mcp = FastMCP("Echo", port=args.port) create_fetch_url_tool(mcp) mcp.run(transport="streamable-http") else: mcp = FastMCP("Echo") create_fetch_url_tool(mcp) mcp.run() # stdio transport ``` -------------------------------- ### Run MCP Server with Streamable HTTP on Custom Port Source: https://github.com/davidshtian/mcp-on-aws-bedrock/blob/hand/README.md Launch the MCP server with the streamable-http transport on a specified custom port. This allows for flexible network configuration. ```bash python fetch_url_mcp_server.py --transport streamable-http --port 8080 ``` -------------------------------- ### Python Agentic Loop Client with Stdio Transport Source: https://context7.com/davidshtian/mcp-on-aws-bedrock/llms.txt Launches an MCP server as a subprocess using stdio, initializes a client session, discovers tools, and runs a multi-turn Bedrock Converse loop. Handles tool use by dispatching requests to the MCP session and feeding results back into the conversation. Exits when the model provides a final text response. ```python import asyncio, json, boto3, logging from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def convert_tool_format(tools): return {"tools": [{"toolSpec": {"name": t.name, "description": t.description, "inputSchema": {"json": t.inputSchema}}} for t in tools]} async def main(): bedrock = boto3.client("bedrock-runtime") # uses ~/.aws/credentials or IAM role async with stdio_client( StdioServerParameters(command="uv", args=["run", "fetch_url_mcp_server.py"]) ) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools_result = await session.list_tools() system = [{"text": "You are a helpful AI assistant. You have access to the following tools: " + json.dumps([{"name": t.name, "description": t.description, "inputSchema": t.inputSchema} for t in tools_result.tools])}] messages = [{"role": "user", "content": [{"text": "Fetch https://www.example.com and summarize it."}]}] while True: response = bedrock.converse( modelId="us.amazon.nova-pro-v1:0", messages=messages, system=system, inferenceConfig={"maxTokens": 300, "topP": 0.1, "temperature": 0.3}, toolConfig=convert_tool_format(tools_result.tools), ) output_message = response["output"]["message"] messages.append(output_message) for content in output_message["content"]: if "text" in content: print("Model:", content["text"]) if response["stopReason"] == "tool_use": for tool_request in output_message["content"]: if "toolUse" in tool_request: tool = tool_request["toolUse"] try: tool_response = await session.call_tool(tool["name"], tool["input"]) tool_result = {"toolUseId": tool["toolUseId"], "content": [{"text": str(tool_response)}]} except Exception as err: tool_result = {"toolUseId": tool["toolUseId"], "content": [{"text": f"Error: {err}"}], "status": "error"} messages.append({"role": "user", "content": [{"toolResult": tool_result}]}) else: break # Model finished; no more tool calls asyncio.run(main()) ``` -------------------------------- ### Convert MCP Tools to Bedrock toolConfig Source: https://context7.com/davidshtian/mcp-on-aws-bedrock/llms.txt Transforms MCP Tool objects into the format required by the Bedrock Converse API. This is necessary as Bedrock does not natively support the MCP format. ```python from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client import boto3, asyncio, json def convert_tool_format(tools): converted_tools = [] for tool in tools: converted_tool = { "toolSpec": { "name": tool.name, "description": tool.description, "inputSchema": {"json": tool.inputSchema}, } } converted_tools.append(converted_tool) return {"tools": converted_tools} ``` ```python # Example: inspect what the converted format looks like async def preview_tool_format(): async with stdio_client( StdioServerParameters(command="uv", args=["run", "fetch_url_mcp_server.py"]) ) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools_result = await session.list_tools() bedrock_tool_config = convert_tool_format(tools_result.tools) print(json.dumps(bedrock_tool_config, indent=2)) # Expected output: # { # "tools": [ # { # "toolSpec": { # "name": "fetch_url", # "description": "Fetches a website and returns its content", # "inputSchema": { # "json": { "properties": { "url": { ... } }, ... } # } # } # } # ] # } ``` ```python asyncio.run(preview_tool_format()) ``` -------------------------------- ### Run MCP Client with Streamable HTTP Source: https://github.com/davidshtian/mcp-on-aws-bedrock/blob/hand/README.md Execute the MCP client configured to communicate via streamable-http. This client is intended for use with a server running the streamable-http transport. ```bash uv run client_streamablehttp.py ``` -------------------------------- ### Agentic Loop Client with Streamable HTTP Source: https://context7.com/davidshtian/mcp-on-aws-bedrock/llms.txt Connects to a remote MCP HTTP server via Streamable HTTP. Ensure the server is running before execution. This client manages the agent's interaction loop with AWS Bedrock, including tool calling and response handling. ```python import asyncio, json, boto3, logging from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def convert_tool_format(tools): return {"tools": [{"toolSpec": {"name": t.name, "description": t.description, "inputSchema": {"json": t.inputSchema}}} for t in tools]} async def main(): # Prerequisites: server must already be running: # python fetch_url_mcp_server.py --transport streamable-http --port 8000 bedrock = boto3.client("bedrock-runtime") async with streamablehttp_client("http://localhost:8000/mcp") as (read_stream, write_stream, _): async with ClientSession(read_stream, write_stream) as session: await session.initialize() tools_result = await session.list_tools() system = [{"text": "You are a helpful AI assistant. Tools available: " + json.dumps([{"name": t.name, "description": t.description, "inputSchema": t.inputSchema} for t in tools_result.tools])}] messages = [{"role": "user", "content": [{"text": "Fetch https://www.example.com and summarize it."}]}] while True: response = bedrock.converse( modelId="us.amazon.nova-pro-v1:0", messages=messages, system=system, inferenceConfig={"maxTokens": 300, "topP": 0.1, "temperature": 0.3}, toolConfig=convert_tool_format(tools_result.tools), ) output_message = response["output"]["message"] messages.append(output_message) for content in output_message["content"]: if "text" in content: print("Model:", content["text"]) if response["stopReason"] == "tool_use": for tool_request in output_message["content"]: if "toolUse" in tool_request: tool = tool_request["toolUse"] try: tool_response = await session.call_tool(tool["name"], tool["input"]) tool_result = {"toolUseId": tool["toolUseId"], "content": [{"text": str(tool_response)}]} except Exception as err: tool_result = {"toolUseId": tool["toolUseId"], "content": [{"text": f"Error: {err}"}], "status": "error"} messages.append({"role": "user", "content": [{"toolResult": tool_result}]}) else: break asyncio.run(main()) # Expected output: # INFO:__main__:Available tools: [{'name': 'fetch_url', ...}] # Model: The example.com page contains a basic HTML placeholder... ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.