Building a Policy-Governed Multi-Agent Financial Research Workflow with Omnigent

import os, sys, subprocess, textwrap, pathlib, getpass
def sh(cmd, **kw):
   """Run a command, and on failure present the ACTUAL error, not simply a code."""
   print("$", " ".be part of(map(str, cmd)))
   p = subprocess.run(cmd, textual content=True, capture_output=True, **kw)
   if p.returncode != 0:
       print(p.stdout or "", p.stderr or "", sep="n")
       elevate RuntimeError(f"Command failed ({p.returncode}): {' '.be part of(map(str, cmd))}")
   return p
WORKDIR = pathlib.Path("/content material/omnigent_tutorial")
WORKDIR.mkdir(mother and father=True, exist_ok=True)
VENV = WORKDIR / ".venv"
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "uv"], test=True)
if not (VENV / "bin" / "python").exists():
   sh(["uv", "venv", "--python", "3.12", str(VENV)])
PY = str(VENV / "bin" / "python")
sh(["uv", "pip", "install", "--python", PY, "-q", "omnigent", "requests"])
OMNI = str(VENV / "bin" / "omnigent")
print("n", subprocess.run([OMNI, "--version"], capture_output=True, textual content=True).stdout.strip())

if not os.environ.get("ANTHROPIC_API_KEY"):
   os.environ["ANTHROPIC_API_KEY"] = getpass.getpass("Anthropic API key: ")
env = os.environ.copy()
env["OMNIGENT_NO_UPDATE_CHECK"] = "1"

(WORKDIR / "agent_tools.py").write_text(textwrap.dedent('''
   """Local instruments uncovered to the Omnigent brokers on this tutorial."""
   import requests
   def get_exchange_rate(base_currency: str, target_currency: str) -> dict:
       """Look up the newest FX charge between two ISO-4217 forex codes."""
       r = requests.get(
           "https://api.frankfurter.app/newest",
           params={"from": base_currency.higher(), "to": target_currency.higher()},
           timeout=10,
       )
       r.raise_for_status()
       information = r.json()
       return {
           "base": base_currency.higher(),
           "goal": target_currency.higher(),
           "charge": information["rates"][target_currency.upper()],
           "date": information["date"],
       }
   def word_count(textual content: str) -> int:
       """Count the phrases in a piece of textual content."""
       return len(textual content.cut up())
'''))

(WORKDIR / "fx_research_lead.yaml").write_text(textwrap.dedent('''
   identify: fx_research_lead
   immediate: |
     You are a monetary analysis lead. For any query about forex
     actions: name get_exchange_rate to fetch the stay charge, then hand
     your draft abstract to the text_auditor sub-agent for a readability and
     size test earlier than giving your ultimate reply to the consumer.
   executor:
     harness: claude-sdk
   instruments:
     get_exchange_rate:
       sort: operate
       callable: agent_tools.get_exchange_rate
     text_auditor:
       sort: agent
       immediate: |
         You audit quick items of monetary writing. Call word_count to
         report its size, flag any unexplained jargon, and recommend one
         concrete readability enchancment.
       instruments:
         word_count:
           sort: operate
           callable: agent_tools.word_count
   insurance policies:
     cap_calls:
       sort: operate
       handler: omnigent.insurance policies.builtins.security.max_tool_calls_per_session
       factory_params:
         restrict: 20
     finances:
       sort: operate
       handler: omnigent.insurance policies.builtins.price.cost_budget
       factory_params:
         max_cost_usd: 1.00
'''))

env["PYTHONPATH"] = str(WORKDIR)
query = (
   "What is the present USD to EUR alternate charge? Give me a two-sentence "
   "abstract I may paste into a shopper notice."
)
outcome = subprocess.run(
   [OMNI, "run", str(WORKDIR / "fx_research_lead.yaml"), "-p", question, "--no-session"],
   cwd=WORKDIR, env=env, stdin=subprocess.DEVNULL,
   capture_output=True, textual content=True, timeout=300,
)
print("n" + "=" * 70)
print(outcome.stdout.strip() or "(no stdout)")
if outcome.returncode != 0 or "error" in outcome.stdout.decrease():
   print("-" * 70)
   print("stderr:", outcome.stderr[-2000:])
   print(f"nDebug: test ~/.omnigent/logs/runner/ , or rerun with:n"
         f"  !{OMNI} --debug --log-to-stderr run {WORKDIR/'fx_research_lead.yaml'} -p "..." --no-session")
print("=" * 70)
print(f"""
Next steps:
 • Explore the CLI:  !{OMNI} run --help
 • Bundled demo brokers:
       !{OMNI} polly -p "evaluation this repo" --no-session
       !{OMNI} debby -p "brainstorm 3 names for a espresso store" --no-session
 • YAML schema:  https://github.com/omnigent-ai/omnigent/blob/principal/docs/AGENT_YAML_SPEC.md
 • Policies:     https://github.com/omnigent-ai/omnigent/blob/principal/docs/POLICIES.md
""")