When one agent isn't enough

Planner agent

class ResearchQuestions(BaseModel):
    questions: list[str] = Field(description="4-6 standalone research questions")

PLANNER_INSTRUCTIONS = (
    "Given a broad research topic, break it into 4-6 specific, "
    "focused sub-questions that together provide a comprehensive answer. "
    "Each question is sent by itself to an isolated researcher. "
    "Researchers cannot see the original topic, other questions, "
    "or other researchers' answers. Therefore every question MUST: "
    "- Be independently answerable with no prior findings... "
    "- Include the relevant subject, location, constraints... ")

planner = Agent(client=build_chat_client(), name="planner",
                instructions=PLANNER_INSTRUCTIONS)
					

Each researcher only sees its own question, so every question has to stand on its own.

📎 orchestrator/agents/planner_agent.py

Fan-out planned questions to research nodes

# Ask the planner agent for questions, as structured output
result = await self.agent.run(payload.topic,
                              options={"response_format": ResearchQuestions})
questions = validate_questions(result.value)

# Tell the collector how many answers to expect
await ctx.send_message(
    ResearchPlan(expected_responses=len(questions), ...),
    target_id="research_collector")

# Send each question to its own researcher node
for i, question in enumerate(questions):
    await ctx.send_message(
        AgentExecutorRequest(messages=[Message("user", [question])]),
        target_id=f"researcher_{i}")
					

📎 orchestrator/agents/workflow.py

Send each question to a sandbox

Each researcher node passes the question directly to a sandbox and returns either the response or an error.

class ResearcherExecutor(Executor):
    @handler
    async def run(self, request: AgentExecutorRequest,
                  ctx: WorkflowContext[AgentExecutorResponse]):
        question = request.messages[-1].text
        try:
            text = await self.run_in_sandbox(question)
        except Exception as ex:
            text = json.dumps({"question": question,
                               "error": str(ex), ...})
        await ctx.send_message(AgentExecutorResponse(...))
					

📎 orchestrator/agents/workflow.py

Start sandbox with research agent and tools

async def run_in_sandbox(question: str) -> str:
    sandbox_id = f"agent-{index}-{uuid.uuid4().hex[:8]}"
    await sandbox_mgr.create_sandbox(sandbox_id, question)        # create
    try:
        while True:                                               # execute
            await asyncio.sleep(2)
            status = await sandbox_mgr.get_status(sandbox_id)
            if status.status == "done":
                result = await sandbox_mgr.get_result(sandbox_id) # collect
                break
    finally:
        await sandbox_mgr.delete_sandbox(sandbox_id)              # delete
    return json.dumps({"question": result.question,
                       "answer": result.answer, ...})
					

📎 sandbox_researcher.py

Research agent inside the sandbox

Runs in each sandbox with Foundry's hosted web search, and returns a structured finding.

class ResearchFinding(BaseModel):
    answer: str
    sources: list[str]
    confidence: float

agent = Agent(
    client=FoundryChatClient(project_client=project_client, model=model),
    name="ResearchAgent",
    instructions="Use web search to find current, factual information, "
                 "then synthesize a comprehensive answer that cites sources.",
    tools=[FoundryChatClient.get_web_search_tool()],
)
response = await agent.run(question, options={"response_format": ResearchFinding})
return response.value.model_dump()
					

📎 research-agent/app.py

Collect answers for each wave

A plain executor: waits for every planned answer, then sends one dossier to the reviewer.

class ResearchCollector(Executor):
    @handler
    async def set_plan(self, plan: ResearchPlan, ctx):
        self.plan = plan
        await self.release_if_ready(ctx)

    @handler
    async def collect_response(self, response: AgentExecutorResponse, ctx):
        self.responses.append(response)
        await self.release_if_ready(ctx)

    async def release_if_ready(self, ctx):
        if len(self.responses) >= self.plan.expected_responses:
            await ctx.send_message(ResearchDossier(findings=..., wave=...))
					

📎 orchestrator/agents/workflow.py

Reviewer agent: approve or research more

class ReviewDecision(BaseModel):
    status: Literal["approved", "needs_more_research"]
    rationale: str
    follow_up_questions: list[str]

result = await self.agent.run(dossier, options={"response_format": ReviewDecision})
decision = validate_review_decision(result.value)
if decision.status == "needs_more_research" and waves_remaining > 0:
    await ctx.send_message(FollowUpResearch(decision.follow_up_questions, ...),
                           target_id="planner")
else:
    await ctx.send_message(ApprovedDossier(...), target_id="report_writer")
					

📎 reviewer_agent.py · 📎 workflow.py

Report writer agent

REPORT_WRITER_INSTRUCTIONS = (
    "Given the original topic, an evidence dossier from multiple research "
    "agents, and the reviewer's assessment, produce a single comprehensive "
    "markdown report. Include an executive summary, key findings organized "
    "by theme, material limitations, and a conclusion. Cite the supplied "
    "sources where available. Do not invent evidence or sources.")

prompt = format_dossier(dossier.topic, dossier.findings) + dossier.review_rationale
result = await self.agent.run(prompt)
await ctx.yield_output(result.text)  # the final report, streamed to the app
					

📎 report_writer_agent.py · 📎 workflow.py