Multi-agent systems

You’ll use team.py. It holds the same pretend network as before, plus a VPN gateway whose certificate has expired. It can run as one agent, or as a supervisor with a server agent and a network agent.

The tools, approvals, hop limit and call counts always behave exactly as shown. Which tools each model picks, and its wording, can differ a little on your laptop. On macOS or Linux, use python3 and nano.

1One agent, two problems

Run this:

cd C:\ai-lab\agent
notepad team.py

Paste this whole block into team.py, then save:

import json
import urllib.request

MODEL = "llama3.2:3b"
QUESTION = "Staff working from home can't open the file share, and nobody in the office can print."
MODE = "single"       # "single" = one agent with every tool, "team" = a supervisor and two specialists
REVIEW = False        # True = a reviewer checks the final answer against the findings
DELEGATE = False      # True = specialists may hand work to each other
APPROVE = True        # True = ask you before any tool that changes something
MAX_STEPS = 6         # model calls allowed per agent
MAX_HOPS = 4          # how deep hand-offs may go
SUPERVISOR_RULES = "You lead an IT support team. Give each part of the problem to the right specialist, then write a short answer for the user from their findings."

# A pretend network, so the lab is safe to run on any laptop
SERVERS = {
    "web01": {"role": "intranet web server", "disk": 41, "services": {"nginx": "running"}},
    "file01": {"role": "file server", "disk": 62, "services": {"smb": "running"}},
    "print01": {"role": "print server for floors 1 to 3", "disk": 58, "services": {"spooler": "stopped"}},
}
VPN = "vpn01: certificate expired on 2026-10-18, remote users cannot connect"
FINDINGS = []         # every tool result, for the reviewer
CALLS = 0             # model calls, to see the cost


def list_servers():
    return "\n".join(f"{name}: {s['role']}" for name, s in SERVERS.items())


def check_service(server, service):
    return f"{service} on {server}: {SERVERS[server]['services'].get(service, 'not installed')}"


def restart_service(server, service):
    SERVERS[server]["services"][service] = "running"
    return f"{service} on {server} restarted, now running"


def ping(host):
    return f"{host}: reachable from the office network, 1 ms"


def check_vpn():
    return VPN


def tool(name, description, **params):
    props = {p: {"type": "string", "description": d} for p, d in params.items()}
    return {"type": "function", "function": {"name": name, "description": description,
            "parameters": {"type": "object", "properties": props, "required": list(params)}}}


TOOLS = {
    "list_servers": tool("list_servers", "List all servers and what each one does."),
    "check_service": tool("check_service", "Show whether a service is running.", server="Server name", service="Service name"),
    "restart_service": tool("restart_service", "Restart a service. Only use it when check_service shows a problem.", server="Server name", service="Service name"),
    "ping": tool("ping", "Check that a host answers on the network.", host="Host name, for example file01"),
    "check_vpn": tool("check_vpn", "Show the state of the VPN gateway that home users connect through."),
    "ask_server_agent": tool("ask_server_agent", "Give a task to the server specialist.", task="What to check or fix"),
    "ask_network_agent": tool("ask_network_agent", "Give a task to the network specialist.", task="What to check or fix"),
}
SERVER_TOOLS = ["list_servers", "check_service", "restart_service"]
NETWORK_TOOLS = ["ping", "check_vpn"]
CHANGES = {"restart_service"}


def chat(messages, tool_names):
    global CALLS
    CALLS += 1
    body = {"model": MODEL, "messages": messages, "stream": False, "options": {"temperature": 0}}
    if tool_names:
        body["tools"] = [TOOLS[t] for t in tool_names]
    req = urllib.request.Request("http://localhost:11434/api/chat", data=json.dumps(body).encode(),
                                 headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req) as r:
        return json.load(r)["message"]


def ask_server_agent(task, depth):
    tools = SERVER_TOOLS + (["ask_network_agent"] if DELEGATE else [])
    return run_agent("server", "You are the server specialist. Use your tools, then report what you found in one sentence.", task, tools, depth)


def ask_network_agent(task, depth):
    tools = NETWORK_TOOLS + (["ask_server_agent"] if DELEGATE else [])
    return run_agent("network", "You are the network specialist. Use your tools, then report what you found in one sentence.", task, tools, depth)


FUNCS = {"list_servers": list_servers, "check_service": check_service, "restart_service": restart_service,
         "ping": ping, "check_vpn": check_vpn, "ask_server_agent": ask_server_agent, "ask_network_agent": ask_network_agent}


def run_agent(name, rules, task, tool_names, depth=0):
    pad = "    " * depth
    if depth > MAX_HOPS:
        return f"Error: too many hand-offs, stopped at {MAX_HOPS}"
    messages = [{"role": "system", "content": rules}, {"role": "user", "content": task}]
    for step in range(1, MAX_STEPS + 1):
        msg = chat(messages, tool_names)
        messages.append(msg)
        if not msg.get("tool_calls"):
            return msg["content"].strip()
        for call in msg["tool_calls"]:
            fn, args = call["function"]["name"], call["function"]["arguments"]
            print(f"{pad}[{name}] {fn}(" + ", ".join(f"{k}={v!r}" for k, v in args.items()) + ")")
            if fn in CHANGES and APPROVE and input(f"{pad}  Allow this change? [y/n] ").strip().lower() != "y":
                result = "The user did not allow this change."
            else:
                try:
                    result = FUNCS[fn](depth=depth + 1, **args) if fn.startswith("ask_") else FUNCS[fn](**args)
                except Exception as e:
                    result = f"Error: {e}"
            if not fn.startswith("ask_"):
                FINDINGS.append(result)
            print(f"{pad}  -> " + result.replace("\n", "\n" + pad + "     "))
            messages.append({"role": "tool", "content": result, "tool_name": fn})
    return f"Stopped: no answer after {MAX_STEPS} steps."


def review(answer):
    rules = "You check answers before they reach users. Every claim must be supported by the findings. Reply APPROVED, or REJECTED: and the reason."
    text = "Findings:\n" + "\n".join(FINDINGS) + "\n\nAnswer:\n" + answer
    return chat([{"role": "system", "content": rules}, {"role": "user", "content": text}], [])["content"].strip()


print(f"Mode: {MODE} | Review: {'on' if REVIEW else 'off'} | Delegation between specialists: {'on' if DELEGATE else 'off'}")
print("Question:", QUESTION)
if MODE == "single":
    answer = run_agent("agent", "You are an IT support agent. Check with the tools before you answer. Answer in two sentences.",
                       QUESTION, SERVER_TOOLS + NETWORK_TOOLS)
else:
    answer = run_agent("supervisor", SUPERVISOR_RULES, QUESTION, ["ask_server_agent", "ask_network_agent"])
print("Answer:", answer)
if REVIEW:
    verdict = review(answer)
    print("[reviewer]", verdict)
    if verdict.startswith("REJECTED"):
        task = f"{QUESTION}\n\nFindings:\n" + "\n".join(FINDINGS) + f"\n\nA reviewer rejected this answer: {answer}\n{verdict}\nRewrite it using only the findings."
        answer = chat([{"role": "system", "content": SUPERVISOR_RULES}, {"role": "user", "content": task}], [])["content"].strip()
        print("Answer (rewritten):", answer)
        print("[reviewer]", review(answer))
print(f"Model calls: {CALLS}")

The question has two problems: home users can’t open the file share, and printing is down. Start with one agent that has every tool. Type y when it asks to restart something.

Run this:

python team.py

You should see something like this (the model’s words and choices may differ a little):

Mode: single | Review: off | Delegation between specialists: off
Question: Staff working from home can't open the file share, and nobody in the office can print.
[agent] list_servers()
  -> web01: intranet web server
     file01: file server
     print01: print server for floors 1 to 3
[agent] check_service(server='print01', service='spooler')
  -> spooler on print01: stopped
[agent] restart_service(server='print01', service='spooler')
  Allow this change? [y/n] y
  -> spooler on print01 restarted, now running
[agent] check_service(server='file01', service='smb')
  -> smb on file01: running
Answer: Printing is fixed: the spooler on print01 had stopped and I restarted it. The file share on file01 is running, so staff at home should be able to open it.
Model calls: 5

It fixed printing, then checked the file share service, saw it running, and said home users should be fine. They aren’t. It had a check_vpn tool and never used it.

One agent with every tool followed the first lead and stopped. That’s the problem a team can solve.

✅ Checkpoint: Model calls: 5, and the answer says nothing about the VPN.

2Turn it into a team

Change this line:

MODE = "team"

Run this:

python team.py

You should see something like this (the model’s words and choices may differ a little):

Mode: team | Review: off | Delegation between specialists: off
Question: Staff working from home can't open the file share, and nobody in the office can print.
[supervisor] ask_network_agent(task="Staff working from home can't open the file share. Find out why.")
    [network] check_vpn()
      -> vpn01: certificate expired on 2026-10-18, remote users cannot connect
  -> The VPN certificate on vpn01 expired on 2026-10-18, so home users can't connect and can't reach the file share.
[supervisor] ask_server_agent(task='Nobody in the office can print. Find out why and fix it.')
    [server] check_service(server='print01', service='spooler')
      -> spooler on print01: stopped
    [server] restart_service(server='print01', service='spooler')
      Allow this change? [y/n] y
      -> spooler on print01 restarted, now running
  -> The spooler on print01 had stopped; I restarted it and it is running now.
Answer: Printing works again: the print spooler on print01 had stopped and has been restarted. Home users can't open the file share because the VPN certificate on vpn01 expired on 18 October; the network team needs to renew it.
Model calls: 7

Read it as a tree. The supervisor gave the file share problem to the network agent, which checked the VPN and found the expired certificate. The server agent fixed printing.

The answer now covers both problems, and it says what nobody could fix: the certificate needs renewing. It took 7 model calls instead of 5.

✅ Checkpoint: The network agent calls check_vpn, the server agent restarts the spooler, and the answer mentions both.

3Add a reviewer

Change this line:

REVIEW = True

Run this:

python team.py

You should see something like this (the model’s words and choices may differ a little):

Mode: team | Review: on | Delegation between specialists: off
Question: Staff working from home can't open the file share, and nobody in the office can print.
[supervisor] ask_network_agent(task="Staff working from home can't open the file share. Find out why.")
    [network] check_vpn()
      -> vpn01: certificate expired on 2026-10-18, remote users cannot connect
  -> The VPN certificate on vpn01 expired on 2026-10-18, so home users can't connect and can't reach the file share.
[supervisor] ask_server_agent(task='Nobody in the office can print. Find out why and fix it.')
    [server] check_service(server='print01', service='spooler')
      -> spooler on print01: stopped
    [server] restart_service(server='print01', service='spooler')
      Allow this change? [y/n] y
      -> spooler on print01 restarted, now running
  -> The spooler on print01 had stopped; I restarted it and it is running now.
Answer: Printing works again: the print spooler on print01 had stopped and has been restarted. Home users can't open the file share because the VPN certificate on vpn01 expired on 18 October; the network team needs to renew it.
[reviewer] APPROVED
Model calls: 8

The reviewer saw only the raw tool results and the final answer. Every claim in the answer matched a finding, so it approved it. That cost one more model call.

✅ Checkpoint: The run ends with [reviewer] APPROVED and Model calls: 8.

4Say no to a change

Run the same thing again, but type n when the server agent asks to restart the spooler:

Run this:

python team.py

You should see something like this (the model’s words and choices may differ a little):

Mode: team | Review: on | Delegation between specialists: off
Question: Staff working from home can't open the file share, and nobody in the office can print.
[supervisor] ask_network_agent(task="Staff working from home can't open the file share. Find out why.")
    [network] check_vpn()
      -> vpn01: certificate expired on 2026-10-18, remote users cannot connect
  -> The VPN certificate on vpn01 expired on 2026-10-18, so home users can't connect and can't reach the file share.
[supervisor] ask_server_agent(task='Nobody in the office can print. Find out why and fix it.')
    [server] check_service(server='print01', service='spooler')
      -> spooler on print01: stopped
    [server] restart_service(server='print01', service='spooler')
      Allow this change? [y/n] n
      -> The user did not allow this change.
  -> The spooler on print01 has stopped; I didn't restart it because the change wasn't approved.
Answer: Printing is down because the print spooler on print01 has stopped; restarting it needs approval, so please ask the IT team. Home users can't open the file share because the VPN certificate on vpn01 expired on 18 October; the network team needs to renew it.
[reviewer] APPROVED
Model calls: 8

The approval check lives in your code, so it works at every level of the team. The server agent reported honestly that it didn’t restart anything, and the supervisor passed that on.

✅ Checkpoint: Nothing is restarted, the answer says the restart needs approval, and the reviewer approves it.