Skip to main content

Cordum + AutoGen: Multi-Agent Tool Partitioning

Multi-agent teams should not hand every participant the same mutation surface. This guide discovers the live bridge catalog once, selects an explicit subset for each AG2 agent, and fails if a named tool is not advertised.

Tool partitioning limits what the model can select. It does not replace Cordum policy or identity: production roles should use separate bridge credentials and server-side scopes as well.

Before you start​

  • Complete the AutoGen quickstart.
  • Bootstrap Cordum with ./tools/scripts/quickstart.sh, then load the secure TLS environment and install the bridge pack from the common framework setup.
  • Put cordum-mcp-bridge on PATH.
  • Export OPENAI_API_KEY.

1. Install​

python -m venv .venv
source .venv/bin/activate
pip install "cordum-adapters[autogen]" "autogen-ext[openai]>=0.4"

2. Partition the advertised catalog​

Save as multi_agent_tools.py:

import asyncio
import os

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient
from cordum_agent_adapters.autogen.modern import build_ag2_tools
from cordum_agent_adapters.mcp_client import McpStdioClient

ROLE_TOOLS = {
    "planner": {"cordum.workflow.run"},
    "operator": {"cordum.workflow.cancel"},
    "reviewer": {"cordum.job.approve", "cordum.job.reject"},
}


def select_tools(tool_defs: list[dict], names: set[str]) -> list[dict]:
    by_name = {
        str(tool["name"]): tool
        for tool in tool_defs
        if isinstance(tool.get("name"), str)
    }
    missing = names - by_name.keys()
    if missing:
        raise RuntimeError(
            f"bridge did not advertise {sorted(missing)}; "
            f"available={sorted(by_name)}"
        )
    return [by_name[name] for name in sorted(names)]


async def main() -> None:
    client = McpStdioClient(
        command=["cordum-mcp-bridge"],
        env=os.environ.copy(),
    )
    model = OpenAIChatCompletionClient(
        model=os.getenv("OPENAI_MODEL", "gpt-4.1-mini"),
        api_key=os.environ["OPENAI_API_KEY"],
    )
    try:
        advertised = client.list_tools()
        agents = []
        for role, names in ROLE_TOOLS.items():
            selected = select_tools(advertised, names)
            native_tools = build_ag2_tools(client, tools=selected)
            agents.append(
                AssistantAgent(
                    name=role,
                    model_client=model,
                    tools=native_tools,
                    system_message=(
                        f"You are the {role}. Use only the tools assigned "
                        "to you and require explicit identifiers."
                    ),
                )
            )
            print(f"{role}: {[tool.name for tool in native_tools]}")

        # Construction proves the team accepts the partitioned agents. Run it
        # only after supplying real workflow, run, and job identifiers.
        RoundRobinGroupChat(
            participants=agents,
            termination_condition=MaxMessageTermination(6),
        )
    finally:
        await model.close()
        client.close()


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

Run it:

python multi_agent_tools.py

The output is derived from the live tools/list response. The script does not claim that an action ran and does not print a made-up conversation.

3. Add a server-side backstop​

The bridge submits each tool call as a Cordum job whose capability is the MCP tool name. A v1 policy can therefore block a capability even if application code accidentally hands it to the wrong agent:

version: v1
rules:
  - id: block-workflow-cancel-in-tutorial
    match:
      capabilities:
        - cordum.workflow.cancel
    decision: deny
    reason: "Cancellation disabled for this tutorial tenant"

Use separate tenants, agent identities, API keys, or bridge processes when roles need different server-side authorization. Do not model authorization as an unforwarded per-call label.

4. Run a real team deliberately​

Replace the construction-only comment with a team.run_stream(...) call only after you have:

  1. A known workflow and run identifier.
  2. A known job currently awaiting approval.
  3. A prompt that states which identifier each agent may use.
  4. A server-side policy and credential scope for each role.

Then inspect the actual tool invocation and policy events in the dashboard. Do not use a static transcript as evidence that the integration ran.

Failure semantics​

  • Approval keeps the bridge job nonterminal and blocks the original tool call. Resolve it in the dashboard before CORDUM_MCP_CALL_TIMEOUT; the same call then continues without an approval-reference retry.
  • A timeout or hard deny returns isError: true; McpStdioClient raises McpToolError, which the AG2 adapter surfaces through its tool-error path.

Next steps​