Agents as Tools Workflow

import asyncio
from fast_agent import FastAgent

fast = FastAgent("Agents-as-Tools simple demo")

@fast.agent(
    name="NY-Project-Manager",
    instruction="Return NY time + timezone, plus a one-line project status.",
    servers=["time"],
)
@fast.agent(
    name="London-Project-Manager",
    instruction="Return London time + timezone, plus a one-line news update.",
    servers=["time"],
)
@fast.agent(
    name="PMO-orchestrator",
    instruction=(
        "Get reports. Always use one tool call per project/news. "  # parallelization
        "Responsibilities: NY projects: [OpenAI, Fast-Agent, Anthropic]. "
        "London news: [Economics, Art, Culture]. "  # routing
        "Aggregate results and add a one-line PMO summary."
    ),
    default=True,
    agents=["NY-Project-Manager", "London-Project-Manager"],  # orchestrator-workers
)
async def main() -> None:
    async with fast.run() as agent:
        await agent("Get PMO report. Projects: all. News: Art, Culture")

if __name__ == "__main__":
    asyncio.run(main())

@fast.agent(
    name="billing_specialist",
    instruction="Handle billing, payments, and invoice questions",
)
@fast.agent(
    name="technical_support",
    instruction="Resolve technical issues and product questions",
)
@fast.agent(
    name="account_manager",
    instruction="Handle account changes and general inquiries",
)
@fast.agent(
    name="customer_service",
    instruction="Route customer inquiries to the appropriate specialist based on the topic.",
    agents=["billing_specialist", "technical_support", "account_manager"],
)

@fast.agent(
    name="translator",
    instruction="Translate text to the specified language",
)
@fast.agent(
    name="translation_coordinator",
    instruction=(
        "Translate the given text to French, German, and Spanish. "
        "Use one tool call per language. Return all translations."
    ),
    agents=["translator"],
)

@fast.agent(
    name="researcher",
    instruction="Research topics using web search",
    servers=["fetch"],
)
@fast.agent(
    name="writer",
    instruction="Write comprehensive articles",
)
@fast.agent(
    name="editor",
    instruction="Edit and improve content",
)
@fast.agent(
    name="content_pipeline",
    instruction=(
        "Create a blog post: "
        "1. Call researcher to gather information "
        "2. Call writer with research to draft article "
        "3. Call editor with draft to polish final version"
    ),
    agents=["researcher", "writer", "editor"],
)

Child agent name: "NY-Project-Manager"
Tool name: "agent__NY-Project-Manager"

{
  "tools": [
    {"name": "get_current_time", "type": "mcp_tool"},
    {"name": "agent__NY-Project-Manager", "type": "agent_tool"},
    {"name": "agent__London-Project-Manager", "type": "agent_tool"}
  ]
}

# Parent makes 3 parallel calls to "reporter" agent
Tool call 1 → reporter[1]  # Independent clone
Tool call 2 → reporter[2]  # Independent clone  
Tool call 3 → reporter[3]  # Independent clone

# After completion, usage merges back to template "reporter" agent

Before parallel execution:
▎▶ Chatting      ▎ PM-orchestrator              sonnet turn 1

During parallel execution:
▎▶ Chatting      ▎ PM-orchestrator              sonnet turn 1
▎▶ Chatting      ▎ NY-Project-Manager[1]        haiku turn 1
▎▶ Calling tool  ▎ NY-Project-Manager[2]        haiku (get_time)
▎▶ Chatting      ▎ London-Project-Manager[1]    haiku turn 1

After completion:
▎▶ Chatting      ▎ PM-orchestrator              sonnet turn 2
▎✓ Ready         ▎ NY-Project-Manager[1]        haiku
▎✓ Ready         ▎ NY-Project-Manager[2]        haiku
▎✓ Ready         ▎ London-Project-Manager[1]    haiku

@fast.agent(
    name="project_status_agent",
    instruction="Get project status including health, blockers, and next steps",
    servers=["jira", "github"],
)
@fast.agent(
    name="dashboard_coordinator",
    instruction=(
        "Generate a status dashboard for all active projects. "
        "Call project_status_agent once per project in parallel. "
        "Projects: [ProjectA, ProjectB, ProjectC, ProjectD]. "
        "Aggregate into a summary dashboard with highlights and concerns."
    ),
    agents=["project_status_agent"],
    max_parallel=10,
)
async def main() -> None:
    async with fast.run() as agent:
        await agent.dashboard_coordinator.send("Generate current project dashboard")

@fast.agent(
    name="summarizer",
    instruction="Create concise summaries of long documents",
)
@fast.agent(
    name="sentiment_analyzer",
    instruction="Analyze sentiment and emotional tone",
)
@fast.agent(
    name="entity_extractor",
    instruction="Extract key entities (people, places, organizations)",
)
@fast.agent(
    name="content_processor",
    instruction=(
        "Process documents by: "
        "1. Determining which analysis types are relevant "
        "2. Calling appropriate agents in parallel "
        "3. Synthesizing a comprehensive report. "
        "Available: summarizer, sentiment_analyzer, entity_extractor"
    ),
    agents=["summarizer", "sentiment_analyzer", "entity_extractor"],
)
async def main() -> None:
    async with fast.run() as agent:
        await agent.content_processor.send("Process customer_feedback.txt")

@fast.agent(
    name="academic_researcher",
    instruction="Search academic papers and journals",
    servers=["fetch"],
)
@fast.agent(
    name="news_researcher",
    instruction="Find recent news and current events",
    servers=["fetch"],
)
@fast.agent(
    name="code_researcher",
    instruction="Search code repositories and documentation",
    servers=["github", "fetch"],
)
@fast.agent(
    name="research_coordinator",
    instruction=(
        "Conduct comprehensive research on the given topic. "
        "Determine which research types are relevant, call those agents, "
        "and synthesize findings into a coherent report. "
        "Available: academic_researcher, news_researcher, code_researcher"
    ),
    agents=["academic_researcher", "news_researcher", "code_researcher"],
    servers=["filesystem"],  # Coordinator can also use tools directly
)
async def main() -> None:
    async with fast.run() as agent:
        await agent.research_coordinator.send(
            "Research the latest developments in transformer architectures"
        )

@fast.agent(
    name="doc_writer",
    instruction="Write technical documentation in the specified language",
)
@fast.agent(
    name="doc_coordinator",
    instruction=(
        "Generate documentation in English, Spanish, French, German, and Japanese. "
        "Call doc_writer once per language with the language specified in the message. "
        "Aggregate all versions and create an index page."
    ),
    agents=["doc_writer"],
    servers=["filesystem"],
    max_parallel=5,
)
async def main() -> None:
    async with fast.run() as agent:
        await agent.doc_coordinator.send(
            "Generate API documentation for the authentication module"
        )

# Automatic default schema:
{
  "type": "object",
  "properties": {
    "message": {
      "type": "string",
      "description": "Message to send to the agent"
    }
  },
  "required": ["message"]
}

from fast_agent.agents.agent_types import AgentConfig

@fast.agent(
    name="data_processor",
    config=AgentConfig(
        instruction="Process data with specified parameters",
        tool_input_schema={
            "type": "object",
            "properties": {
                "data_source": {"type": "string"},
                "operation": {"type": "string", "enum": ["filter", "transform", "aggregate"]},
                "parameters": {"type": "object"}
            },
            "required": ["data_source", "operation"]
        }
    ),
)

@fast.agent(
    name="orchestrator",
    instruction="Coordinate tasks across child agents",
    agents=["worker1", "worker2"],
    history_source="orchestrator",      # Children see parent's history
    history_merge_target="orchestrator", # Children's history merges back to parent
)

import logging
logging.basicConfig(level=logging.DEBUG)

# You'll see:
# DEBUG: Spawning detached instance: NY-Project-Manager[1]
# DEBUG: Child tool call: agent__NY-Project-Manager[1] with args {"message": "..."}
# DEBUG: Instance NY-Project-Manager[1] completed, merging usage
# DEBUG: Shutting down detached instance: NY-Project-Manager[1]