Multi-Agent Legal Research Swarm
Orchestrate multiple AI agents that collaborate on complex legal research tasks using the Moonlit API as their knowledge backbone.
What you'll build
Complex legal research often requires multiple specialized skills: finding relevant case law, analyzing statutory provisions, comparing across jurisdictions, and synthesizing findings into a structured memo. A multi-agent system delegates these tasks to specialized agents that work in parallel and collaborate on the final output. This tutorial builds a research swarm with three specialized agents: a Search Agent that formulates and executes Moonlit API queries, a Document Agent that retrieves and summarizes full texts, and a Synthesis Agent that produces the final research memo with proper citations. The orchestrator manages the workflow, passing context between agents and merging their outputs. Each agent uses the Moonlit API through different endpoints depending on its role, demonstrating how the API serves as a shared knowledge backbone for AI-powered legal workflows.
Architecture
┌──────────────────┐
│ Orchestrator │
│ (task planner) │
└────────┬─────────┘
│
┌──────────┼──────────────┐
▼ ▼ ▼
┌─────────┐ ┌─────────┐ ┌────────────┐
│ Search │ │ Document│ │ Synthesis │
│ Agent │ │ Agent │ │ Agent │
└────┬────┘ └────┬────┘ └──────┬─────┘
│ │ │
└──────────┼────────────┘
▼
┌───────────────────────────┐
│ Moonlit API │
│ (shared knowledge base) │
└───────────────────────────┘Prerequisites
- A Moonlit API key
- Python 3.11+
- Anthropic API key
- pip install requests anthropic
Step-by-step
Define the agent base class
Create an abstract agent with access to the Moonlit API and an LLM. Each specialized agent overrides the execute method with its specific behavior.
import anthropic
import requests
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
MOONLIT_KEY = "your-api-key"
BASE_URL = "https://api.moonlit.ai/v1.1"
@dataclass
class AgentContext:
question: str
jurisdictions: list[str]
search_results: list[dict] = field(default_factory=list)
documents: list[dict] = field(default_factory=list)
analysis: str = ""
class LegalAgent(ABC):
def __init__(self, name: str, system_prompt: str):
self.name = name
self.system_prompt = system_prompt
self.client = anthropic.Anthropic()
def llm(self, prompt: str) -> str:
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=4096,
system=self.system_prompt,
messages=[{"role": "user", "content": prompt}],
)
return response.content[0].text
def search(self, query: str, jurisdictions: list[str], n: int = 10) -> list[dict]:
response = requests.post(
f"{BASE_URL}/search/hybrid_search_reranked",
headers={
"Ocp-Apim-Subscription-Key": MOONLIT_KEY,
"Content-Type": "application/json",
},
json={"query": query, "jurisdictions": jurisdictions, "num_results": n, "semantic_weight": 0.6},
)
return response.json()["result"]["results"]
def retrieve(self, doc_id: str) -> dict:
response = requests.get(
f"{BASE_URL}/document/retrieve_document",
headers={"Ocp-Apim-Subscription-Key": MOONLIT_KEY},
params={"DocumentIdentifier": doc_id},
)
return response.json()
@abstractmethod
def execute(self, ctx: AgentContext) -> AgentContext:
passImplement the specialized agents
Build three agents: Search Agent formulates queries and finds relevant documents, Document Agent retrieves and summarizes full texts, and Synthesis Agent writes the final memo.
class SearchAgent(LegalAgent):
def __init__(self):
super().__init__("SearchAgent", "You formulate precise legal search queries.")
def execute(self, ctx: AgentContext) -> AgentContext:
strategies = self.llm(
f"Generate 3 different search queries for: {ctx.question}\n"
f"Jurisdictions: {ctx.jurisdictions}\nReturn one query per line."
).strip().split("\n")
all_results, seen_ids = [], set()
for query in strategies[:3]:
results = self.search(query.strip(), ctx.jurisdictions, n=10)
for r in results:
if r["identifier"] not in seen_ids:
all_results.append(r)
seen_ids.add(r["identifier"])
ctx.search_results = all_results
print(f"[SearchAgent] Found {len(all_results)} unique documents")
return ctx
class DocumentAgent(LegalAgent):
def __init__(self):
super().__init__("DocumentAgent", "You analyze legal documents and extract key holdings.")
def execute(self, ctx: AgentContext) -> AgentContext:
docs = []
for result in ctx.search_results[:8]:
doc = self.retrieve(result["identifier"])
summary = self.llm(
f"Summarize the key legal holding in 2-3 sentences:\n\n"
f"Title: {doc.get('title', '')}\n"
f"Text: {doc.get('text', doc.get('summary', ''))[:3000]}"
)
doc["agent_summary"] = summary
docs.append(doc)
ctx.documents = docs
print(f"[DocumentAgent] Analyzed {len(docs)} documents")
return ctx
class SynthesisAgent(LegalAgent):
def __init__(self):
super().__init__("SynthesisAgent",
"You write professional legal research memos with proper citations.")
def execute(self, ctx: AgentContext) -> AgentContext:
sources = "\n\n".join(
f"[{d['identifier']}] {d.get('title','')}: {d.get('agent_summary','')}"
for d in ctx.documents
)
memo = self.llm(
f"Write a research memo answering: {ctx.question}\n\n"
f"Use these sources (cite by ECLI):\n{sources}"
)
ctx.analysis = memo
print(f"[SynthesisAgent] Memo generated ({len(memo)} chars)")
return ctxBuild the orchestrator
The orchestrator coordinates the agent pipeline, passing context from one agent to the next and managing the overall research workflow.
class ResearchOrchestrator:
def __init__(self):
self.agents = [SearchAgent(), DocumentAgent(), SynthesisAgent()]
def research(self, question: str, jurisdictions: list[str]) -> str:
ctx = AgentContext(question=question, jurisdictions=jurisdictions)
for agent in self.agents:
print(f"\n--- Running {agent.name} ---")
ctx = agent.execute(ctx)
return ctx.analysis
# Run it
orchestrator = ResearchOrchestrator()
memo = orchestrator.research(
question="How do EU member states implement the right to explanation "
"for automated decision-making under GDPR Article 22?",
jurisdictions=["European Union", "Netherlands", "Germany"],
)
print("\n" + "=" * 60)
print(memo)