> ## Documentation Index
> Fetch the complete documentation index at: https://docs.agno.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Team-in-Step Executor HITL

> Demonstrates executor-level HITL when a Team is used inside a workflow Step.

Demonstrates executor-level HITL when a Team is used inside a workflow Step. The team's member agent has a tool with `requires_confirmation=True`. The pause propagates: member agent -> team -> step -> workflow.

```python team_in_step.py theme={null}
"""
Team-in-Step Executor HITL
============================

Demonstrates executor-level HITL when a Team is used inside a workflow Step.
The team's member agent has a tool with `requires_confirmation=True`.
The pause propagates: member agent -> team -> step -> workflow.

Usage:
    .venvs/demo/bin/python cookbook/04_workflows/08_human_in_the_loop/executor_hitl/03_team_in_step.py
"""

from agno.agent import Agent
from agno.db.postgres import PostgresDb
from agno.models.openai import OpenAIChat
from agno.team import Team
from agno.tools import tool
from agno.workflow.step import Step
from agno.workflow.types import StepInput, StepOutput
from agno.workflow.workflow import Workflow
from rich.console import Console
from rich.prompt import Prompt

console = Console()

db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai")


@tool(requires_confirmation=True)
def get_the_weather(city: str) -> str:
    """Get the current weather for a city.

    Args:
        city: The city to get weather for.
    """
    return f"It is currently 70 degrees and cloudy in {city}"


weather_agent = Agent(
    name="WeatherAgent",
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[get_the_weather],
    db=db,
    telemetry=False,
)

weather_team = Team(
    name="WeatherTeam",
    model=OpenAIChat(id="gpt-4o-mini"),
    members=[weather_agent],
    db=db,
    telemetry=False,
    instructions=[
        "You MUST delegate all weather-related tasks to the WeatherAgent.",
        "Do NOT try to answer weather questions yourself.",
    ],
)


def save_result(step_input: StepInput) -> StepOutput:
    prev = step_input.previous_step_content or "no previous content"
    return StepOutput(content=f"Result saved: {prev}")


workflow = Workflow(
    name="TeamWeatherWorkflow",
    db=db,
    steps=[
        Step(name="team_weather", team=weather_team),
        Step(name="save", executor=save_result),
    ],
    telemetry=False,
)

# ---------------------------------------------------------------------------
# Run
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    response = workflow.run("What is the weather in Tokyo?")

    if response.is_paused and response.step_requirements:
        for step_req in response.step_requirements:
            if step_req.requires_executor_input:
                console.print(
                    f"[bold yellow]Workflow paused at step '{step_req.step_name}'[/]\n"
                    f"Executor: [bold cyan]{step_req.executor_name}[/] "
                    f"(type: {step_req.executor_type})"
                )

                for executor_req in step_req.executor_requirements or []:
                    tool_exec = (
                        executor_req.get("tool_execution", {})
                        if isinstance(executor_req, dict)
                        else getattr(executor_req, "tool_execution", None)
                    )
                    if tool_exec:
                        tool_name = (
                            tool_exec.get("tool_name", "?")
                            if isinstance(tool_exec, dict)
                            else getattr(tool_exec, "tool_name", "?")
                        )
                        tool_args = (
                            tool_exec.get("tool_args", {})
                            if isinstance(tool_exec, dict)
                            else getattr(tool_exec, "tool_args", {})
                        )
                        console.print(f"  Tool: [bold blue]{tool_name}({tool_args})[/]")

                answer = (
                    Prompt.ask("Approve?", choices=["y", "n"], default="y")
                    .strip()
                    .lower()
                )

                for executor_req in step_req.executor_requirements or []:
                    if isinstance(executor_req, dict):
                        executor_req["confirmation"] = answer == "y"
                        if (
                            "tool_execution" in executor_req
                            and executor_req["tool_execution"]
                        ):
                            executor_req["tool_execution"]["confirmed"] = answer == "y"
                    else:
                        if answer == "y":
                            executor_req.confirm()
                        else:
                            executor_req.reject(note="User declined")

        response = workflow.continue_run(response)

    console.print(f"\n[bold green]Final output:[/] {response.content}")
```

## Run the Example

<Steps>
  <Snippet file="create-venv-step.mdx" />

  <Step title="Install dependencies">
    ```bash theme={null}
    uv pip install -U agno fastapi openai psycopg-binary sqlalchemy
    ```
  </Step>

  <Step title="Export your OpenAI API key">
    <CodeGroup>
      ```bash Mac/Linux theme={null}
      export OPENAI_API_KEY="your_openai_api_key_here"
      ```

      ```bash Windows theme={null}
      $Env:OPENAI_API_KEY="your_openai_api_key_here"
      ```
    </CodeGroup>
  </Step>

  <Snippet file="run-pgvector-step.mdx" />

  <Step title="Run the example">
    Save the code above as `team_in_step.py`, then run:

    ```bash theme={null}
    python team_in_step.py
    ```
  </Step>
</Steps>

Full source: [cookbook/04\_workflows/08\_human\_in\_the\_loop/executor\_hitl/03\_team\_in\_step.py](https://github.com/agno-agi/agno/blob/main/cookbook/04_workflows/08_human_in_the_loop/executor_hitl/03_team_in_step.py)