Harnessing the power of Strands Agents SDK

import json
import random
from strands import Agent, tool
from strands.models.ollama import OllamaModel

@tool
def get_current_weather(location: str) -> dict:
    """
    Get the current weather for a given location.

    Args:
        location (str): The location to get the weather for.

    Returns:
        dict: A dictionary containing the current weather information.
    """
    return {
        "location": location,
        "temperature": random.randint(60, 110),
        "condition": random.choice(["Sunny", "Cloudy", "Rainy", "Windy"])
    }

# Let's use Ollama
def make_model() -> OllamaModel:
    return OllamaModel(
        host=OLLAMA_HOST,
        model_id=OLLAMA_MODEL_ID,
        temperature=0.6,
    )

agent = Agent(
    name="Weatherman",
    description="A helpful assistant that provides current weather information for a given location.",
    system_prompt=(
        "You are a helpful assistant that provides current weather information for a given location. "
        "You have access to a tool called 'get_current_weather' that takes a location as input and returns the current weather information. "
        "When a user asks for the weather, you should use this tool to get the information and provide it in your response."
        "Always use the tool to get the weather information instead of making up the data yourself. Be friendly and concise in your responses."
    ),
    model=make_model(),
    tools=[get_current_weather]
)

result = agent("What's the weather like in New York City?")

print("\n" + "=" * 60)
print("CONVERSATION HISTORY:")
print("=" * 60)
print(json.dumps(agent.messages, indent=2, default=str))

from strands.tools.mcp import MCPClient
from mcp.client.streamable_http import streamablehttp_client

local_mcp_filtered = MCPClient(
    lambda: streamablehttp_client("http://127.0.0.1:8081/mcp"),
    tool_filters={
        "allowed": ["suv__search_ddgs"]
    }
)

research_agent = Agent(
    name="ResearchAgent",
    model=make_model(),
    system_prompt="""
    You are a research assistant. You will be given a research question and you will need to use the tools available to you and return the receieved 
    response as is. Only provide what you find from the tools, do not provide any additional information or make any assumptions.
    """,
    tools=[local_mcp_filtered]
)

orchestrator_agent = Agent(
    name="OrchestratorAgent",
    model=make_model(),
    system_prompt="""
    You are a technical writer. You will be given a research question and you will need to find relevant information to answer the question.
    You can use the tools available to you to search for information. You should provide a summary of your findings and any relevant links or references.
    """,
    tools=[research_agent]
)

research_question = "What is the current state of research on the use of AI in logistics and supply chain management?"
response = orchestrator_agent(research_question)

session_manager = FileSessionManager(
    session_id="user-session-001",
    storage_dir="./basics/sessions",
)

agent = Agent(
    name="UserPreferenceAgent",
    model=make_model(),
    session_manager=session_manager,
    conversation_manager=SlidingWindowConversationManager(window_size=5),
    system_prompt="""
    You are a helpful assistant that answers user queries keeping in mind their preferences.
    You will use the 'get_user_preferences' tool to fetch user preferences based on their userid.
    If you do not have it, always ask the user for their userid before providing any recommendations or answers.
    Do not make assumptions about the user's preferences without using the tool to fetch them first.
    Do not provide any recommendations or answers without first confirming the user's preferences using the tool.
    """,
    tools=[get_user_preferences],
)

while True:
    user_input = input("User: ")
    if user_input.lower() in ["exit", "quit"]:
        break

    response = agent(user_input)
    print(f"Agent: {response}")

from strands.vended_plugins.steering import (
    SteeringHandler, LLMSteeringHandler, Proceed, Guide, ToolSteeringAction, LedgerProvider,
)

class SteerDrinkHandler(SteeringHandler):
    """
    A steering handler that enforces the workflow for serving drinks.
    It ensures that the agent first gets the ingredients needed, checks the ingredients, and takes the order before serving the drink.
    """
    name = "mixer-workflow"

    def __init__(self):
        super().__init__(context_providers=[LedgerProvider()])

    async def steer_before_tool(self, *, agent, tool_use, **kwargs) -> ToolSteeringAction:
        print(f"[STEERING] Evaluating tool call: {tool_use.get('name')}")

        ledger = self.steering_context.data.get("ledger", {})
        tool_calls = ledger.get("tool_calls", [])

        if tool_use.get("name") != "serve_drink":
            # Must have a get_ingredient_needed call before serving a drink
            if not any(call.get("name") == "take_order" for call in tool_calls):
                return Guide(
                    reason="Cannot serve drink without taking the order first.",
                    guidance="Please take the order before serving the drink."
                )

        elif tool_use.get("name") != "take_order":
            # Must have a get_ingredient_needed call before serving a drink
            if not any(call.get("name") == "get_ingredient_needed" for call in tool_calls):
                return Guide(
                    reason="Cannot serve drink without getting the ingredients needed first.",
                    guidance="Please get the ingredients needed before serving the drink."
                )
    
            # Must have a check_ingredients call before serving a drink
            if not any(call.get("name") == "check_ingredients" for call in tool_calls):
                return Guide(
                    reason="Cannot serve drink without checking ingredients first.",
                    guidance="Please check the ingredients availability before serving the drink."
                )
        
        return Proceed(reason="Continue with the next step.")

class ToneGuardrailHandler(LLMSteeringHandler):
    """
    A steering handler that evaluates the tone of the agent's responses and provides guidance if the tone is not appropriate.
    It checks for negative language, blaming the customer, jargon, assumptions, and sarcasm in the agent's responses.
    """
    name = "tone-guardrail"

    def __init__(self):
        super().__init__(
            model=make_model(),
            system_prompt="""
            Evaluate the drink order conversation for tone and adherence to customer service guidelines.
            If the conversation violates any of the following guidelines, provide specific guidance on how to correct it:
            - Avoid using negative language
            - Avoid using sarcasm or humor that may be misinterpreted
            If violated, provide specific guidance on what to fix.
            """
        )

    async def steer_after_model(self, **kwargs):
        print("[TONE] Evaluating agent response...")
        result = await super().steer_after_model(**kwargs)
        action_type = type(result).__name__
        print(f"[TONE] {'Approved' if action_type == 'Proceed' else 'Guided: ' + getattr(result, 'reason', '')}")
        return result

agent = Agent(
    name="DrinkMixerAgent",
    description="A helpful assistant that can take drink orders and serve drinks.",
    model=make_model(),
    tools=[
        check_ingredients,
        get_ingredient_needed,
        take_order,
        serve_drink
    ],
    plugins=[
        SteerDrinkHandler(),
        tone_handler
    ],
    system_prompt="""
    You are a helpful assistant that can take drink orders, check for ingredients, and serve drinks.
    You have access to the following tools:
    - get_ingredient_needed: Get the ingredients needed for a specific drink.
    - check_ingredients: Check if the required ingredients are available.
    - take_order: Take an order for a specific drink.
    - serve_drink: Serve the prepared drink.
    When a user places an order, you should follow these steps:
    - First get the ingredients needed for the drink using tool call
    - Then check if the ingredients are available using tool call
    - If all ingredients are available, take the order using tool call
    - Finally, serve the drink using tool call
    You will not serve the drink if any of the ingredients are unavailable and let the user know which ingredients are missing.
    Always follow this workflow and use the tools to perform these actions instead of making up the data yourself. Be friendly and concise in your responses.
    """,
)

result = agent("I want to order an Old Fashioned.")

class PaymentApprovalHook(HookProvider):
    """
    Intercept Payment and require approval before proceeding with the payment.
    """
    def register_hooks(self, registry: HookRegistry) -> None:
        registry.add_callback(BeforeToolCallEvent, self.validate_payment_approval)

    def validate_payment_approval(self, event: BeforeToolCallEvent):
        if event.tool_use["name"] != "finalize_payment":
            return

        approval = event.interrupt(
            "payment-approval",
            reason={"amount": event.tool_use["input"]["amount"]}
        )

        if approval.lower() != "y":
            event.cancel_tool = "User denied payment. Payment not processed."

 agent = Agent(
    name="SupplyAgent",
    description="A helpful assistant that can order supplies and make payments.",
    system_prompt=(
        "You are a helpful assistant that can order supplies and make payments. "
        "You have access to two tools: 'order_supplies' and 'finalize_payment'. "
        "To order supplies, you should use the 'order_supplies' tool to place the order and provide the order confirmation and price in your response. "
        "To process a payment, you should use the 'finalize_payment' tool to process the payment and provide the payment confirmation and status in your response. "
        "If user does not approve the payment, you should not proceed. Do not ask additional questions. "
        "Always use the tools to perform these actions instead of making up the data yourself. Be friendly and concise in your responses."
    ),
    model=make_model(),
    tools=[order_supplies, finalize_payment],
    hooks=[PaymentApprovalHook()]
)
  
result = agent("I want to order 10 boxes of pens and 5 reams of paper. Please place the order and process the payment.")
while result.stop_reason == "interrupt":
    for interrupt in result.interrupts:
        approval = input(f"\n❓ Approve payment of '{interrupt.reason['amount']}'? (y/N): ")
        result = agent([
            {
                "interruptResponse": {
                    "interruptId": interrupt.id,
                    "response": approval,
                }
            }
        ])

---
name: fortune
description: Fortune-telling skill
allowed-tools: None
---
# Fortune-Telling Skill

You are a fortune-telling expert. When asked to provide a fortune:

1. Generate a random fortune that is positive and uplifting.
2. Ensure the fortune is concise and easy to understand.
3. Avoid any negative or harmful predictions.

from strands.vended_plugins.skills import AgentSkills

lugin = AgentSkills(
    skills=['./skills/'],
)

agent = Agent(
    name="FortuneTellerAgent",
    model=make_model(),
    plugins=[plugin]
)

result = agent("Provide a fortune for the user.")
print(f"Fortune: {result}")

from strands.multiagent import GraphBuilder

order_ingestor_agent = Agent() # This is used to take an order
carrier_selector_agent = Agent() # Only task for this is to redirect to procing agents
fedex_price_negotiator_agent = Agent() # Get pricing information from FedEx
dhl_price_negotiator_agent = Agent() # DHL
ups_price_negotiator_agent = Agent() # UPS, what else
shipment_finalizer_agent = Agent() # Finalize on the cheapest and despatch

# Main code starts here
shipment_graph = GraphBuilder()

shipment_graph.add_node(order_ingestor_agent, "order_ingestor_agent")
shipment_graph.add_node(carrier_selector_agent, "carrier_selector_agent")
shipment_graph.add_node(fedex_price_negotiator_agent, "fedex_price_negotiator_agent")
shipment_graph.add_node(dhl_price_negotiator_agent, "dhl_price_negotiator_agent")
shipment_graph.add_node(ups_price_negotiator_agent, "ups_price_negotiator_agent")
shipment_graph.add_node(shipment_finalizer_agent, "shipment_finalizer_agent")

shipment_graph.add_edge("order_ingestor_agent", "carrier_selector_agent")
shipment_graph.add_edge("carrier_selector_agent", "fedex_price_negotiator_agent")
shipment_graph.add_edge("carrier_selector_agent", "dhl_price_negotiator_agent")
shipment_graph.add_edge("carrier_selector_agent", "ups_price_negotiator_agent")
shipment_graph.add_edge("fedex_price_negotiator_agent", "shipment_finalizer_agent")
shipment_graph.add_edge("dhl_price_negotiator_agent", "shipment_finalizer_agent")
shipment_graph.add_edge("ups_price_negotiator_agent", "shipment_finalizer_agent")

shipment_graph.set_execution_timeout(600)
graph = shipment_graph.build()

result = graph("Send a 20x12 inch item weighing 5 pounds to Boston in the most economical way possible")

print(f"Status: {result.status}")
print(f"Execution order: {[n.node_id for n in result.execution_order]}")

from strands_tools import workflow

def orchestrate_workflow() -> str:
    """
    Orchestrates the workflow using the workflow module, allowing for more complex interactions between agents.
    """
    orchestrator = Agent(
        name="OrchestratorAgent",
        model=model,
        tools=[workflow]
    )
    
    # Define the workflow steps
    orchestrator.tool.workflow(
        action="create",
        workflow_id="coffee_pipeline",
        tasks=[
            {
                "task_id": "get-coffee-beans",
                "description": "Get coffee beans",
                "system_prompt": "You are a coffee bean specialist. Get coffee beans.",
                "priority": 5,  # if two tasks are same priority, they run in parallel.
            },
            {
                "task_id": "get-french-press",
                "description": "Get a French press",
                "system_prompt": "You have all brewing equipments. Get a French press from stock.",
                "priority": 5,
            },
            {
                "task_id": "grind-coffee-beans",
                "description": "Grind the coffee beans",
                "system_prompt": "You are a coffee grinding specialist. Grind the coffee beans for brewing in a french press.",
                "priority": 4,
            },
            {
                "task_id": "brew-coffee",
                "description": "Brew the coffee",
                "system_prompt": "You are a coffee brewing specialist. Brew the coffee using the French press and the ground coffee beans.",
                "priority": 3,
            },
            {
                "task_id": "serve-coffee",
                "description": "Serve the coffee",
                "system_prompt": "You are a coffee serving specialist. Serve the brewed coffee to the user.",
                "priority": 2,
            },
        ]
    )
    
    orchestrator.tool.workflow(action="start", workflow_id="coffee_pipeline")
    result = orchestrator.tool.workflow(action="status", workflow_id="coffee_pipeline")
    print(f"\n{result['content'][0]['text']}")
    return result['status']

from strands.vended_plugins.goal import GoalLoop

excite = GoalLoop(
  goal="""
    Make the sentence fun and catchy. Fill everything with excitement and energy. Make it sound like a thrilling adventure. 
    Use vivid imagery and descriptive language to create a sense of wonder and awe. Make it sound like a once-in-a-lifetime 
    experience. Make it sound like a story that will be remembered for years to come. Write in 50 words or less.
  """,
  max_attempts=5,
  timeout=60,
)

agent = Agent(
    name="FortuneTellerAgent",
    model=make_model(),
    plugins=[excite]
)

result = agent("Why is the sky blue?")
print('\n' + '-' * 60)
print(f"{excite.last_result(agent)}")

from strands.multiagent import Swarm

COMMON_PROMPT = {
    "system": (
        "\n\nSwarm ground rules:\n"
        "- You all work in same medical research lab. You are all highly skilled and experienced in your respective fields.\n"
        "- You are all working together to solve a complex problem. You must collaborate and communicate effectively to achieve your goal.\n"
        "- Stay strictly inside your own specialty; do not repeat points that has been already made.\n"
        "- Keep each turn short and concrete: About 3 bullet points max"
        "- If another specialist would clearly add more value on the current topic, hand off the conversation to them. Do not try to answer questions outside your specialty.\n"
        "- If you are unsure about something, ask for clarification or hand off to another specialist.\n"
        "- Do NOT hand off back and forth on the same point. Once you and the other specialist have discussed a point, do not hand off back to them on the same point. Move on to the next point or hand off to a different specialist.\n"
        "- Never ask a user for any information. There is no user in this scenario. You are all working together to solve a problem.\n"
    )
}

def make_agents() -> list[Agent]:
    model = make_model()
    
    principal_investigator = Agent(
        name="Principal Investigator",
        description="You are the Principal Investigator of a medical research lab. You are responsible for overseeing the research projects and ensuring that they are conducted ethically and effectively.",
       	....
      
    lab_manager = Agent(
        name="Lab Manager",
        description="You are the Lab Manager of a medical research lab. You are responsible for managing the day-to-day operations of the lab",
        ....
      
    bioinformatics_specialist = Agent(
        name="Bioinformatics Specialist",
        description="You are a Bioinformatics Specialist in a medical research lab. You are responsible for analyzing complex biological data using computational tools and techniques.",
        ....
      
    warehouse_manager = Agent(
        name="Warehouse Manager",
        description="You are the Warehouse Manager of a medical research lab. You are responsible for managing the lab's inventory, including ordering supplies, tracking stock levels, and ensuring that materials are stored and handled properly.",
        ....
      
    translation_specialist = Agent(
        name="Translation Specialist",
        description="You are a Translation Specialist in a medical research lab. You are responsible for translating scientific documents, research papers, and communications between team members who speak different languages.",
        ....
      
    return [
        principal_investigator,
        lab_manager,
        bioinformatics_specialist,
        warehouse_manager,
        translation_specialist,
    ]
      
def build_scribe() -> Agent:
    """A single, non-swarm agent that writes the final answer once."""
    return Agent(
        name="synthesizer",
        system_prompt=(
            "You are a scribe working at the research lab. You will be given the "
            "raw brainstorm notes produced by five specialists. "
            "Merge their input into ONE cohesive, "
            "non-redundant answer to the original topic, organized under "
            "clear headings. Resolve any overlaps or contradictions "
            "yourself instead of listing both. Keep it actionable."
        ),
        model=make_model(),
    )
      
def run_brainstorm(topic: str) -> None:
    agents = make_agents()

    swarm = Swarm(
        agents,
        entry_point=agents[0],  # principal_investigator starts
        max_handoffs=MAX_HANDOFFS,
        max_iterations=MAX_ITERATIONS,
        execution_timeout=EXECUTION_TIMEOUT_SECONDS,
        node_timeout=NODE_TIMEOUT_SECONDS,
        repetitive_handoff_detection_window=REPETITIVE_HANDOFF_WINDOW,
        repetitive_handoff_min_unique_agents=REPETITIVE_HANDOFF_MIN_UNIQUE,
    )

    task = (
        f"Lab topic to brainstorm: {topic}\n\n"
        "Each specialist should contribute concrete ideas from their own "
        "area of expertise. Hand off to whichever specialist is best "
        "suited for the next open question, and stop handing off once the "
        "topic has been covered from all five angles."
    )

    print(f"\n=== Running swarm brainstorm on: {topic!r} ===\n")
    result = swarm(task)  # single call — this call itself does not loop

    # ---- Report exactly why/how the swarm stopped (exit criteria hit) ----
    print(f"Swarm finished with status: {result.status}")
    print(f"Agents involved, in order: "
          f"{[node.node_id for node in result.node_history]}")
    print(f"Total agent turns used: {result.execution_count} "
          f"(cap was {MAX_ITERATIONS})")
    print(f"Wall-clock time: {result.execution_time} ms "
          f"(cap was {EXECUTION_TIMEOUT_SECONDS * 1000:.0f} ms)\n")

    if result.status != "COMPLETED":
        print(
            "Note: the swarm stopped due to a safety limit "
            "(timeout / handoff cap / repetitive-handoff detection) "
            "rather than natural completion. Partial notes below are "
            "still used for synthesis.\n"
        )

    # ---- Collect each specialist's contribution for synthesis ----------
    notes = []
    for node in result.node_history:
        node_result = result.results.get(node.node_id)
        if node_result is not None:
            notes.append(f"### {node.node_id}\n{node_result.result}")
    combined_notes = "\n\n".join(notes)

    # ---- One final, single (non-looping) synthesis call ----------------
    synthesizer = build_scribe()
    final_answer = synthesizer(
        f"Original topic: {topic}\n\n"
        f"Specialist brainstorm notes:\n\n{combined_notes}\n\n"
        "Write the final combined answer now."
    )

    print("=== FINAL SYNTHESIZED ANSWER ===\n")
    print(final_answer)

if __name__ == "__main__":
    if len(sys.argv) > 1:
        topic_arg = " ".join(sys.argv[1:])
    else:
        topic_arg = (
            "We have to find a new medicine for influenza that is effective against all strains, "
            "has minimal side effects, and can be produced at scale. "
            "We should be able to send the medicine all over the world, so documentation and translation will be important. " \
            "We need to consider the logistics of storing and distributing the medicine, as well as the regulatory requirements for approval in different countries."
            "What are the key considerations and steps we need to take to achieve this goal?"
        )
    run_brainstorm(topic_arg)