Une seule politique d'approbation pour tous les sous-agents

import asyncio

from mistralai import agents

PUBLISHED: list[str] = []


@agents.tool
def run_tests(feature: str) -> str:
    """Run the test suite of a feature."""
    return f"{feature}: 128 passed"


@agents.tool
def publish_docs(feature: str, text: str) -> str:
    """Publish the docs of a feature."""
    PUBLISHED.append(feature)
    return f"published the docs of {feature}"


lead = agents.Agent(
    model="mistral-large-4",
    name="release-lead",
    instructions=(
        "You ship features. For each feature, start one tester and one docs writer, "
        "wait for both, then say in one line what was done."
    ),
    subagents=[
        agents.Subagent(
            name="tester",
            description="Runs the tests of one feature and reports the result.",
            instructions="Call run_tests once, then answer with the result.",
            tools=[run_tests],
            builtins=agents.Builtins(subagents=False),
        ),
        agents.Subagent(
            name="docs-writer",
            description="Writes two lines of docs for one feature and publishes them.",
            instructions="Write two lines of docs, call publish_docs once, then answer in one line.",
            tools=[publish_docs],
            builtins=agents.Builtins(subagents=False),
        ),
    ],
    permission_mode="ask",  # the default "auto" asks nothing, so no handler runs
)

def ci_policy(
    ctx: agents.ApprovalCallbackContext, request: agents.ApprovalRequest
) -> agents.AllowApproval | agents.KeepCallbackOpen:
    ids = ("client.run_tests", "client.publish_docs", "agents.spawn")  # the children's tools, and starting them
    if request.tool.id in ids:
        return agents.AllowApproval()
    return agents.KeepCallbackOpen()

async def ship(lead: agents.Agent) -> agents.TurnState:
    session = await lead.start_session(on_approval=ci_policy)  # answers the whole tree
    try:
        state = await session.run("Ship feature X", timeout=600)  # children's approvals too
        for child in await session.subagents():  # they can still run after the lead's turn
            print(f"  child session {child.id}")
        return state
    finally:
        await session.cancel()

async def keep_answering(lead: agents.Agent, session_id: str) -> None:
    """A review bot: answers for a session that it does not drive."""
    session = lead.session(session_id, on_approval=ci_policy)
    async for event in session.iter_events():  # replays from the start, then follows
        if isinstance(event, agents.TurnStarted):
            await session.turn(event.turn_id).result()  # answers the whole tree while it waits

async def main() -> None:
    state = await ship(lead)
    print(f"ship: {type(state).__name__}: {state.text}")

    # Another client starts a turn with no handler; the review bot answers its approvals.
    session = await lead.start_session()
    bot = asyncio.create_task(keep_answering(lead, session.id))
    try:
        await session.send_message("Ship feature Y")
        async for event in session.iter_events():
            if isinstance(event, agents.CallbackOpened):
                print(f"  open: {event.callback.title}")
            if isinstance(event, agents.TurnEnded) and event.session_id == session.id:
                print(f"bot: {type(event.state).__name__}: {event.state.text}")
                break
    finally:
        bot.cancel()
        await session.cancel()
    print(f"docs published: {PUBLISHED}")


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