test

Author: Unshure

test

@@ -0,0 +1,14 @@
+from fastmcp import FastMCP
+import time
+
+app = FastMCP()
+
+@app.tool("say")
+def say(input: str) -> str:
+    """Says something."""
+    time.sleep(5)
+    return f"{input}!"
+
+if __name__ == "__main__":
+    app.run(transport="http", host="localhost", port=8080)
+    # app.run()

test-mcp.py

@@ -0,0 +1,36 @@
+from datetime import timedelta
+from strands.tools.mcp import MCPClient
+from mcp.client.streamable_http import streamablehttp_client
+from strands import Agent
+import logging
+import asyncio
+
+logging.getLogger("strands.tools.mcp").setLevel(logging.DEBUG)
+
+# Add a handler to see the logs
+logging.basicConfig(
+    format="%(levelname)s | %(name)s | %(message)s", 
+    handlers=[logging.StreamHandler()]
+)
+
+client = MCPClient(lambda: streamablehttp_client(
+    sse_read_timeout=2,
+    url="http://localhost:8080/mcp"))
+
+with client as mcp:
+    try:
+        # mcp.call_tool_sync(tool_use_id="123", name="say", arguments={"input": "Hello!"})
+        agent = Agent(tools=[mcp.list_tools_sync()])
+        while True:
+            
+            agent(input("\nInput: "))
+    except Exception as e:
+        print(f"\n\n\n\n\nHERE: {e}")
+    
+    # async def call_say_tool():
+    #     await mcp.call_tool_async(tool_use_id="123", name="say", arguments={"input": "Hello!"})
+
+    # asyncio.run(call_say_tool())    
+    # agent = Agent(tools=[mcp.list_tools_sync()])
+    # agent("Use the say tool three times in one request to say the world \"Hello\" \"there\" \"world!\"")
+    # print(agent.messages)