-
Notifications
You must be signed in to change notification settings - Fork 35
Expand file tree
/
Copy pathmain.py
More file actions
60 lines (45 loc) · 1.4 KB
/
main.py
File metadata and controls
60 lines (45 loc) · 1.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
import os
from autogen_agentchat.agents import AssistantAgent
from autogen_core.tools import FunctionTool
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from model.load import load_model
{{#unless isVpc}}
from mcp_client.client import get_streamable_http_mcp_tools
{{/unless}}
app = BedrockAgentCoreApp()
log = app.logger
# Define a simple function tool
def add_numbers(a: int, b: int) -> int:
"""Return the sum of two numbers"""
return a + b
add_numbers_tool = FunctionTool(
add_numbers, description="Return the sum of two numbers"
)
# Define a collection of tools used by the model
tools = [add_numbers_tool]
@app.entrypoint
async def invoke(payload, context):
log.info("Invoking Agent.....")
{{#unless isVpc}}
# Get MCP Tools
mcp_tools = await get_streamable_http_mcp_tools()
{{/unless}}
# Define an AssistantAgent with the model and tools
agent = AssistantAgent(
name="{{ name }}",
model_client=load_model(),
{{#unless isVpc}}
tools=tools + mcp_tools,
{{else}}
tools=tools,
{{/unless}}
system_message="You are a helpful assistant. Use tools when appropriate.",
)
# Process the user prompt
prompt = payload.get("prompt", "What can you help me with?")
# Run the agent
result = await agent.run(task=prompt)
# Return result
return {"result": result.messages[-1].content}
if __name__ == "__main__":
app.run()