Beyond Large Language Models: Mastering Agentic Workflows with the New Open-Action Protocol

{
  "open_action": "1.0",
  "id": "uuid-v4",
  "timestamp": "2026-03-09T13:45:12.345Z",
  "sender": "agent:research_assistant",
  "receiver": "agent:web_searcher",
  "action": {
    "type": "search_web",
    "parameters": {
      "query": "latest breakthroughs in quantum error correction",
      "top_k": 5,
      "lang": "en"
    },
    "metadata": {
      "trace_id": "trace-12345",
      "priority": "high",
      "deadline": "2026-03-09T14:00:00Z"
    }
  },
  "security": {
    "scope": "read:web",
    "signature": "sha256:..."
  }
}

{
  "open_action": "1.0",
  "id": "uuid-v4",
  "timestamp": "...",
  "sender": "agent:web_searcher",
  "receiver": "agent:research_assistant",
  "result": {
    "status": "success",
    "output": [
      {"title": "...", "url": "...", "snippet": "..."},
      ...
    ],
    "log": "Fetched 5 results from Bing API."
  },
  "security": {...}
}

class ResearchOrchestrator:
    def __init__(self, agent_registry):
        self.registry = agent_registry   # maps action.type → endpoint URL
        self.state = {}
    
    async def run(self, goal):
        # 1️⃣ Generate initial search request via LLM
        search_msg = await self._llm_generate_search(goal)
        results = await self._dispatch(search_msg)

        # 2️⃣ Fetch PDFs for top results
        pdf_msgs = [self._make_fetch_msg(r["url"]) for r in results["output"]]
        pdf_keys = await self._parallel_dispatch(pdf_msgs)

        # 3️⃣ Extract text, summarize, and cite
        summary = await self._summarize_and_cite(pdf_keys)

        # 4️⃣ Store final output
        await self._store_summary(summary)
        return summary

# Create a virtual environment
python -m venv .venv
source .venv/bin/activate

# Install required packages
pip install fastapi uvicorn httpx openai pydantic python-dotenv \
            sqlalchemy psycopg2-binary minio

# oap_models.py
from pydantic import BaseModel, Field, Json
from typing import Any, Dict, List, Optional
from uuid import uuid4
from datetime import datetime

class Action(BaseModel):
    type: str
    parameters: Dict[str, Any]
    metadata: Optional[Dict[str, Any]] = None

class Security(BaseModel):
    scope: str
    signature: Optional[str] = None   # In production you would sign with HMAC

class OpenActionMessage(BaseModel):
    open_action: str = "1.0"
    id: str = Field(default_factory=lambda: str(uuid4()))
    timestamp: str = Field(default_factory=lambda: datetime.utcnow().isoformat()+"Z")
    sender: str
    receiver: str
    action: Optional[Action] = None
    result: Optional[Dict[str, Any]] = None
    security: Optional[Security] = None

# registry.py
ACTION_REGISTRY = {
    "search_web": "http://localhost:8000/agent/web_searcher",
    "fetch_pdf": "http://localhost:8000/agent/pdf_fetcher",
    "extract_text": "http://localhost:8000/agent/text_extractor",
    "summarize": "http://localhost:8000/agent/summarizer",
    "cite_papers": "http://localhost:8000/agent/citation_manager",
    "store_summary": "http://localhost:8000/agent/db_writer",
}

# agents.py
import httpx
from fastapi import FastAPI, Request
from oap_models import OpenActionMessage
import os

app = FastAPI()

BING_ENDPOINT = "https://api.bing.microsoft.com/v7.0/search"
BING_KEY = os.getenv("BING_API_KEY")

@app.post("/agent/web_searcher")
async def web_searcher(req: Request):
    msg: OpenActionMessage = await req.json()
    query = msg.action.parameters["query"]
    top_k = msg.action.parameters.get("top_k", 5)

    headers = {"Ocp-Apim-Subscription-Key": BING_KEY}
    params = {"q": query, "count": top_k, "mkt": "en-US"}

    async with httpx.AsyncClient() as client:
        resp = await client.get(BING_ENDPOINT, headers=headers, params=params)
        resp.raise_for_status()
        data = resp.json()

    # Extract a simplified list of results
    results = [
        {"title": r["name"], "url": r["url"], "snippet": r.get("snippet", "")}
        for r in data.get("webPages", {}).get("value", [])
    ]

    # Build Result message
    result_msg = OpenActionMessage(
        sender="agent:web_searcher",
        receiver=msg.sender,
        result={"status": "success", "output": results, "log": "Bing search completed"},
        security=msg.security,
    )
    return result_msg.dict()

from minio import Minio
from pathlib import Path

minio_client = Minio(
    endpoint=os.getenv("MINIO_ENDPOINT"),
    access_key=os.getenv("MINIO_ACCESS_KEY"),
    secret_key=os.getenv("MINIO_SECRET_KEY"),
    secure=False,
)

BUCKET = "pdf-research"

@app.post("/agent/pdf_fetcher")
async def pdf_fetcher(req: Request):
    msg: OpenActionMessage = await req.json()
    url = msg.action.parameters["url"]
    bucket = msg.action.parameters.get("bucket", BUCKET)

    # Ensure bucket exists
    if not minio_client.bucket_exists(bucket):
        minio_client.make_bucket(bucket)

    # Download PDF (simplified, no error handling)
    async with httpx.AsyncClient() as client:
        r = await client.get(url)
        r.raise_for_status()
        data = r.content

    object_name = f"{msg.id}.pdf"
    minio_client.put_object(bucket, object_name, data=bytes(data), length=len(data), content_type="application/pdf")

    result_msg = OpenActionMessage(
        sender="agent:pdf_fetcher",
        receiver=msg.sender,
        result={"status": "success", "object_key": f"{bucket}/{object_name}", "log": "PDF stored"},
        security=msg.security,
    )
    return result_msg.dict()

from pdfminer.high_level import extract_text
import io

@app.post("/agent/text_extractor")
async def text_extractor(req: Request):
    msg: OpenActionMessage = await req.json()
    bucket, obj = msg.action.parameters["object_key"].split("/", 1)
    response = minio_client.get_object(bucket, obj)
    pdf_bytes = response.read()
    text = extract_text(io.BytesIO(pdf_bytes))

    result_msg = OpenActionMessage(
        sender="agent:text_extractor",
        receiver=msg.sender,
        result={"status": "success", "text": text[:5000], "log": "Extracted plain text"},
        security=msg.security,
    )
    return result_msg.dict()

import openai

openai.api_key = os.getenv("OPENAI_API_KEY")

@app.post("/agent/summarizer")
async def summarizer(req: Request):
    msg: OpenActionMessage = await req.json()
    text = msg.action.parameters["text"]
    max_tokens = msg.action.parameters.get("max_tokens", 800)

    response = await openai.ChatCompletion.acreate(
        model="gpt-4o-mini",
        messages=[{"role": "system", "content": "Summarize concisely for an executive audience."},
                  {"role": "user", "content": text}],
        max_tokens=max_tokens,
        temperature=0.3,
    )
    summary = response.choices[0].message.content.strip()

    result_msg = OpenActionMessage(
        sender="agent:summarizer",
        receiver=msg.sender,
        result={"status": "success", "summary": summary, "log": "LLM summarization complete"},
        security=msg.security,
    )
    return result_msg.dict()

import httpx

CROSSREF_API = "https://api.crossref.org/works/"

@app.post("/agent/citation_manager")
async def citation_manager(req: Request):
    msg: OpenActionMessage = await req.json()
    dois = msg.action.parameters["dois"]
    citations = []

    async with httpx.AsyncClient() as client:
        for doi in dois:
            r = await client.get(f"{CROSSREF_API}{doi}")
            if r.status_code == 200:
                data = r.json()["message"]
                citation = f"{data['author'][0]['family']}, {data['author'][0]['given']} ({data['published']['date-parts'][0][0]}). {data['title'][0]}."
                citations.append(citation)

    result_msg = OpenActionMessage(
        sender="agent:citation_manager",
        receiver=msg.sender,
        result={"status": "success", "citations": citations, "log": "Resolved DOIs"},
        security=msg.security,
    )
    return result_msg.dict()

from sqlalchemy import create_engine, Table, Column, String, Text, MetaData
engine = create_engine(os.getenv("POSTGRES_URL"))
metadata = MetaData()
summaries = Table(
    "summaries",
    metadata,
    Column("id", String, primary_key=True),
    Column("title", String),
    Column("content", Text),
    Column("metadata", Text),
)
metadata.create_all(engine)

@app.post("/agent/db_writer")
async def db_writer(req: Request):
    msg: OpenActionMessage = await req.json()
    payload = msg.action.parameters
    with engine.begin() as conn:
        conn.execute(
            summaries.insert().values(
                id=msg.id,
                title=payload["title"],
                content=payload["content"],
                metadata=str(payload.get("metadata", {})),
            )
        )
    result_msg = OpenActionMessage(
        sender="agent:db_writer",
        receiver=msg.sender,
        result={"status": "success", "log": "Stored summary"},
        security=msg.security,
    )
    return result_msg.dict()

# orchestrator.py
import asyncio, httpx, json, uuid, os
from datetime import datetime, timezone
from oap_models import OpenActionMessage, Action, Security

REGISTRY = {
    "search_web": "http://localhost:8000/agent/web_searcher",
    "fetch_pdf": "http://localhost:8000/agent/pdf_fetcher",
    "extract_text": "http://localhost:8000/agent/text_extractor",
    "summarize": "http://localhost:8000/agent/summarizer",
    "cite_papers": "http://localhost:8000/agent/citation_manager",
    "store_summary": "http://localhost:8000/agent/db_writer",
}

SECURITY = Security(scope="full", signature=None)   # placeholder

async def dispatch(action_type: str, params: dict, sender: str, receiver: str):
    msg = OpenActionMessage(
        sender=sender,
        receiver=receiver,
        action=Action(type=action_type, parameters=params),
        security=SECURITY,
    )
    async with httpx.AsyncClient() as client:
        resp = await client.post(REGISTRY[action_type], json=msg.dict())
        resp.raise_for_status()
        return OpenActionMessage(**resp.json())

async def run_research(goal: str):
    # 1️⃣ Generate a search query via the LLM (hard‑coded for brevity)
    search_msg = await dispatch(
        "search_web",
        {"query": goal, "top_k": 3, "lang": "en"},
        sender="orchestrator",
        receiver="agent:web_searcher",
    )
    urls = [r["url"] for r in search_msg.result["output"]]

    # 2️⃣ Parallel PDF fetch & extraction
    fetch_tasks = [
        dispatch("fetch_pdf", {"url": u}, "orchestrator", "agent:pdf_fetcher")
        for u in urls
    ]
    fetch_results = await asyncio.gather(*fetch_tasks)

    extract_tasks = [
        dispatch(
            "extract_text",
            {"object_key": fr.result["object_key"]},
            "orchestrator",
            "agent:text_extractor",
        )
        for fr in fetch_results
    ]
    texts = [er.result["text"] for er in await asyncio.gather(*extract_tasks)]

    # 3️⃣ Summarize each text, then combine
    summary_tasks = [
        dispatch(
            "summarize",
            {"text": t, "max_tokens": 500},
            "orchestrator",
            "agent:summarizer",
        )
        for t in texts
    ]
    summaries = [s.result["summary"] for s in await asyncio.gather(*summary_tasks)]

    combined = "\n\n".join(summaries)

    # 4️⃣ Extract DOIs (simple regex) and ask citation manager
    import re
    dois = re.findall(r"10.\d{4,9}/[-._;()/:A-Z0-9]+", combined, flags=re.I)
    citation_msg = await dispatch(
        "cite_papers",
        {"dois": list(set(dois))},
        "orchestrator",
        "agent:citation_manager",
    )

    final_content = f"{combined}\n\nReferences:\n" + "\n".join(citation_msg.result["citations"])

    # 5️⃣ Store the final summary
    await dispatch(
        "store_summary",
        {"title": f"Research brief on {goal}", "content": final_content, "metadata": {"source_urls": urls}},
        "orchestrator",
        "agent:db_writer",
    )
    print("✅ Research workflow completed.")
    return final_content

if __name__ == "__main__":
    asyncio.run(run_research("latest breakthroughs in quantum error correction"))

# 1️⃣ Start the FastAPI server (agents)
uvicorn agents:app --host 0.0.0.0 --port 8000

# 2️⃣ In another terminal, launch the orchestrator
python orchestrator.py