Build a Multi-Agent System in Python (and When Not To)

import asyncio
import sys

from promptise import build_agent
from promptise.config import StdioServerSpec
from promptise.cross_agent import CrossAgent


async def main():
    # A specialist is an ordinary agent with its own tools.
    billing = await build_agent(
        model="openai:gpt-5-mini",
        servers={"billing": StdioServerSpec(command=sys.executable, args=["billing_server.py"])},
        instructions="You are the billing specialist. Answer from your tools only.",
    )
    # The coordinator gets the specialist as a tool.
    coordinator = await build_agent(
        model="openai:gpt-5-mini",
        servers={},
        instructions="You are a support coordinator. Ask the billing specialist about accounts.",
        cross_agents={
            "billing": CrossAgent(agent=billing, description="Looks up one customer's account."),
        },
    )
    try:
        print("Tools:", coordinator.tool_names)
        result = await coordinator.ainvoke(
            {"messages": [{"role": "user", "content": "What plan is customer acme on?"}]}
        )
        print(result["messages"][-1].content)
    finally:
        await coordinator.shutdown()
        await billing.shutdown()


asyncio.run(main())

pip install promptise
export OPENAI_API_KEY="sk-..."

from promptise.mcp.server import MCPServer

server = MCPServer("billing")

# A stand-in for your billing system.
ACCOUNTS = {
    "acme": {"plan": "Team", "seats": 12, "price_per_seat": 15.00, "renews": "2026-11-01"},
    "globex": {"plan": "Starter", "seats": 3, "price_per_seat": 8.00, "renews": "2026-10-20"},
}
INVOICES = {
    "acme": [
        {"invoice_id": "INV-2041", "date": "2026-10-01", "amount": 180.00, "status": "paid"},
        {"invoice_id": "INV-1987", "date": "2026-09-01", "amount": 165.00, "status": "paid"},
    ],
    "globex": [
        {"invoice_id": "INV-2044", "date": "2026-10-01", "amount": 24.00, "status": "overdue"},
    ],
}


@server.tool()
async def get_account(customer_id: str) -> dict:
    """Get a customer's plan, number of seats, price per seat in USD and renewal date.

    Args:
        customer_id: The customer's ID, for example "acme".
    """
    account = ACCOUNTS.get(customer_id.strip().lower())
    if account is None:
        return {"error": f"No customer with ID {customer_id}."}
    return {"customer_id": customer_id.strip().lower(), **account}


@server.tool()
async def list_invoices(customer_id: str) -> list[dict]:
    """List a customer's invoices, newest first, with amount in USD and status.

    Args:
        customer_id: The customer's ID, for example "acme".
    """
    return INVOICES.get(customer_id.strip().lower(), [])


if __name__ == "__main__":
    server.run()

from promptise.mcp.server import MCPServer

server = MCPServer("docs")

# A stand-in for your help center or knowledge base.
ARTICLES = [
    {
        "title": "Adding seats",
        "text": "Admins add seats under Settings > Billing > Seats. New seats are billed "
        "immediately, prorated to the end of the current billing period.",
    },
    {
        "title": "Changing your plan",
        "text": "Upgrades take effect immediately and are prorated. Downgrades take effect "
        "at the next renewal date. Plans: Starter (8 USD per seat, up to 5 seats), "
        "Team (15 USD per seat), Business (25 USD per seat, includes SSO).",
    },
    {
        "title": "Overdue invoices",
        "text": "If an invoice is more than 14 days overdue, the workspace becomes read-only "
        "until it is paid. Pay under Settings > Billing > Invoices.",
    },
    {
        "title": "Single sign-on (SSO)",
        "text": "SAML SSO is available on the Business plan. Set it up under Settings > Security.",
    },
]


@server.tool()
async def search_help_center(query: str) -> list[dict]:
    """Search the help center and return matching articles with their full text.

    Args:
        query: A few keywords, for example "seats" or "SSO".
    """
    words = [w for w in query.lower().split() if len(w) > 2]
    return [a for a in ARTICLES if any(w in (a["title"] + " " + a["text"]).lower() for w in words)]


if __name__ == "__main__":
    server.run()

    billing = await build_agent(
        model="openai:gpt-5-mini",
        servers={"billing": StdioServerSpec(command=sys.executable, args=["billing_server.py"])},
        instructions=(
            "You are the billing specialist. Answer in a few short lines, with facts "
            "from your tools only. Say so if your tools have no answer."
        ),
    )
    docs = await build_agent(
        model="openai:gpt-5-mini",
        servers={"docs": StdioServerSpec(command=sys.executable, args=["docs_server.py"])},
        instructions=(
            "You are the help center specialist. Answer in a few short lines, from help "
            "center articles only, and name the article you used."
        ),
    )

    coordinator = await build_agent(
        model="openai:gpt-5-mini",
        servers={},
        instructions=(
            "You are a customer support coordinator. You can't see customer data or the "
            "help center yourself, so delegate. Ask the billing specialist about one "
            "customer's account. Ask the docs specialist how the product works, and what "
            "each plan includes and costs. Ask only the specialists the question needs, "
            "with short, specific questions. When it needs both, ask both in the same "
            "step, then combine their answers into one short reply. Don't offer actions "
            "nobody on the team can take."
        ),
        cross_agents={
            "billing": CrossAgent(
                agent=billing,
                description=(
                    "Looks up one customer's account: their plan, seats, what they pay "
                    "today, and their invoices. Doesn't know other plans' prices."
                ),
            ),
            "docs": CrossAgent(
                agent=docs,
                description=(
                    "Answers from the help center: how-to steps, billing policies, and "
                    "what every plan includes and costs per seat."
                ),
            ),
        },
    )

from collections import Counter

from langchain_core.callbacks import BaseCallbackHandler


class TeamTrace(BaseCallbackHandler):
    """Print who asked whom, and count model calls and tokens per agent."""

    def __init__(self, coordinator: str = "coordinator"):
        self.coordinator = coordinator
        self.agent_of_run: dict = {}  # tool run id -> the agent that run belongs to
        self.calls: Counter = Counter()
        self.tokens: Counter = Counter()

    def _agent(self, parent_run_id) -> str:
        return self.agent_of_run.get(parent_run_id, self.coordinator)

    def on_tool_start(self, serialized, input_str, *, run_id, parent_run_id=None, inputs=None, **kwargs):
        name = serialized.get("name", "?")
        caller = self._agent(parent_run_id)
        indent = "" if caller == self.coordinator else "    "
        if name.startswith("ask_agent_"):
            peer = name.removeprefix("ask_agent_")
            self.agent_of_run[run_id] = peer
            print(f"{caller} -> {peer}: {inputs or input_str}")
        else:
            self.agent_of_run[run_id] = caller
            print(f"{indent}{caller} calls {name} {inputs or input_str}")

    def on_tool_end(self, output, *, run_id, parent_run_id=None, **kwargs):
        agent = self.agent_of_run.get(run_id)
        caller = self._agent(parent_run_id)
        if agent != caller:  # a specialist just answered the agent that asked it
            text = str(getattr(output, "content", output))
            print(f"{agent} -> {caller}: {text[:300]}{' …' if len(text) > 300 else ''}")

    def on_llm_end(self, response, *, parent_run_id=None, **kwargs):
        agent = self._agent(parent_run_id)
        usage = response.generations[0][0].message.usage_metadata or {}
        self.calls[agent] += 1
        self.tokens[agent] += usage.get("total_tokens", 0)

    def summary(self) -> str:
        rows = [f"{a}: {self.calls[a]} model calls, {self.tokens[a]:,} tokens" for a in self.calls]
        total = f"total: {sum(self.calls.values())} model calls, {sum(self.tokens.values()):,} tokens"
        return "\n".join(rows + [total])

python support_team.py

import asyncio
import sys
import time

from promptise import build_agent
from promptise.config import StdioServerSpec
from promptise.cross_agent import CrossAgent

from team_trace import TeamTrace

QUESTIONS = [
    "Does customer globex have any unpaid invoices?",
    "Customer acme wants SSO. What would their monthly bill be, and how do they set it up?",
]


async def main():
    # Two specialists, each with its own MCP server and its own instructions.
    billing = await build_agent(
        model="openai:gpt-5-mini",
        servers={"billing": StdioServerSpec(command=sys.executable, args=["billing_server.py"])},
        instructions=(
            "You are the billing specialist. Answer in a few short lines, with facts "
            "from your tools only. Say so if your tools have no answer."
        ),
    )
    docs = await build_agent(
        model="openai:gpt-5-mini",
        servers={"docs": StdioServerSpec(command=sys.executable, args=["docs_server.py"])},
        instructions=(
            "You are the help center specialist. Answer in a few short lines, from help "
            "center articles only, and name the article you used."
        ),
    )

    # The coordinator has no tools of its own, only its two colleagues.
    coordinator = await build_agent(
        model="openai:gpt-5-mini",
        servers={},
        instructions=(
            "You are a customer support coordinator. You can't see customer data or the "
            "help center yourself, so delegate. Ask the billing specialist about one "
            "customer's account. Ask the docs specialist how the product works, and what "
            "each plan includes and costs. Ask only the specialists the question needs, "
            "with short, specific questions. When it needs both, ask both in the same "
            "step, then combine their answers into one short reply. Don't offer actions "
            "nobody on the team can take."
        ),
        cross_agents={
            "billing": CrossAgent(
                agent=billing,
                description=(
                    "Looks up one customer's account: their plan, seats, what they pay "
                    "today, and their invoices. Doesn't know other plans' prices."
                ),
            ),
            "docs": CrossAgent(
                agent=docs,
                description=(
                    "Answers from the help center: how-to steps, billing policies, and "
                    "what every plan includes and costs per seat."
                ),
            ),
        },
    )

    try:
        print("Coordinator tools:", coordinator.tool_names)
        for question in QUESTIONS:
            print(f"\n=== {question}")
            trace = TeamTrace()
            started = time.monotonic()
            result = await coordinator.ainvoke(
                {"messages": [{"role": "user", "content": question}]},
                config={"callbacks": [trace]},
            )
            print(">>>", result["messages"][-1].content)
            print(f"--- {time.monotonic() - started:.1f}s")
            print(trace.summary())
    finally:
        await coordinator.shutdown()
        await billing.shutdown()
        await docs.shutdown()


asyncio.run(main())

async def build_single():
    agent = await build_agent(
        model="openai:gpt-5-mini",
        servers={"billing": BILLING, "docs": DOCS},
        instructions=(
            "You are a customer support assistant. Use the billing tools for a customer's "
            "plan, seats, prices and invoices, and the help center for how the product "
            "works and what each plan includes and costs. Answer in one short reply."
        ),
    )
    return agent, [agent]

    billing = await build_agent(
        model="openai:gpt-5-mini",
        servers={"billing": StdioServerSpec(command=sys.executable, args=["billing_server.py"])},
        instructions="You are the billing specialist.",
        max_invocation_time=1,  # far too short, on purpose
    )

        extra_tools=make_cross_agent_tools(
            {"billing": CrossAgent(agent=billing, description="Looks up one customer's account.")},
            include_broadcast=False,
        ),

import sys

from promptise import build_agent
from promptise.config import StdioServerSpec
from promptise.mcp.server import MCPServer

server = MCPServer("billing-team")
specialist = {}


@server.on_startup
async def start_specialist():
    specialist["agent"] = await build_agent(
        model="openai:gpt-5-mini",
        servers={"billing": StdioServerSpec(command=sys.executable, args=["billing_server.py"])},
        instructions=(
            "You are the billing specialist. Answer in a few short lines, with facts "
            "from your tools only. Say so if your tools have no answer."
        ),
    )


@server.on_shutdown
async def stop_specialist():
    await specialist["agent"].shutdown()


@server.tool()
async def ask_billing_specialist(question: str) -> str:
    """Ask the billing specialist about one customer's account: their plan, seats,
    what they pay today, and their invoices. It doesn't know other plans' prices.

    Args:
        question: One short, specific question that names the customer.
    """
    result = await specialist["agent"].ainvoke({"messages": [{"role": "user", "content": question}]})
    return result["messages"][-1].content


if __name__ == "__main__":
    server.run(transport="http", host="127.0.0.1", port=8260)

    coordinator = await build_agent(
        model="openai:gpt-5-mini",
        # The billing specialist runs in another process, behind an MCP server.
        servers={"billing_team": HTTPServerSpec(url="http://127.0.0.1:8260/mcp")},
        instructions=(
            "You are a customer support coordinator. Delegate: ask the billing specialist "
            "about one customer's account, and the docs specialist how the product works "
            "and what each plan includes and costs. When a question needs both, ask both "
            "in the same step, then combine their answers into one short reply."
        ),
        cross_agents={
            "docs": CrossAgent(
                agent=docs,
                description=(
                    "Answers from the help center: how-to steps, billing policies, and "
                    "what every plan includes and costs per seat."
                ),
            ),
        },
    )