Every primitive. Every pattern. The mechanics behind "build a pipeline, turn it into a tool, repeat."
Install the package, set one environment variable, and run your first agent. No boilerplate, no configuration files.
# Install from PyPI pip install lazybridge # Set your provider key (pick one — or more) export ANTHROPIC_API_KEY="sk-ant-..." export OPENAI_API_KEY="sk-..." export GOOGLE_API_KEY="..."
The simplest possible usage. One agent, one call, one string back. Change the provider string to switch between Anthropic, OpenAI, Google, or DeepSeek.
from lazybridge import LazyAgent ai = LazyAgent("anthropic") answer = ai.text("What is the capital of France?") print(answer) # "Paris" # Full response with token counts resp = ai.chat("Explain quantum entanglement in one sentence.") print(resp.content) print(resp.usage.input_tokens, resp.usage.output_tokens) # Switch provider by changing one string ai_openai = LazyAgent("openai") ai_google = LazyAgent("google") ai_deepseek = LazyAgent("deepseek")
Wrap any Python function as a tool. loop() handles the full call cycle: detect intent → call your function → feed result back → repeat until done.
from lazybridge import LazyAgent, LazyTool def get_weather(city: str) -> str: """Get current weather for a city.""" return f"Weather in {city}: 22°C, partly cloudy" weather_tool = LazyTool.from_function(get_weather) ai = LazyAgent("anthropic") result = ai.loop( "What's the weather like in Rome and Paris?", tools=[weather_tool], ) print(result.content) # loop() automatically: # 1. Sends the message to the LLM # 2. Detects tool call requests # 3. Runs get_weather() with the LLM's arguments # 4. Feeds the result back — repeats until done
Two agents, one session. The researcher's output is injected into the writer's context via LazyContext. Full event log and shared state included.
from lazybridge import LazyAgent, LazySession, LazyContext # Shared container: tracking, store, graph sess = LazySession() researcher = LazyAgent("anthropic", name="researcher", session=sess) writer = LazyAgent("openai", name="writer", session=sess) # Step 1: researcher works independently researcher.loop("Find the top 3 developments in AI this week.") # Step 2: writer reads researcher's output lazily writer.chat( "Write a short newsletter section from this research.", context=LazyContext.from_agent(researcher), ) print(writer.result) # the newsletter text print(sess.events.get()) # full event log print(sess.store.read_all()) # shared state
LazyBridge has five primitives. Each has one job. Understanding what each one is — and what it is not — is the entire mental model.
The single entry point for all LLM interaction. Stateless or stateful. Wraps a provider, a system prompt, and optional tools. Everything starts here.
The unit of composition. Wraps a Python function or an agent. Has a name, description, and JSON Schema. Can be passed to any agent. Can be saved, loaded, nested.
The multi-agent container. Holds a shared store, event log, and graph. Composes N agents into a parallel or chain pipeline with one call.
Lazy system prompt injection. Evaluated at call time, not construction time. Reads agent outputs, store values, or arbitrary functions.
A shared key-value blackboard. In-memory or SQLite-backed. For intentionally decoupled agents that run at different times or in different processes.
The single entry point for all LLM interaction. Stateless by default. Add session=sess to participate in shared state and tracking.
from lazybridge import LazyAgent from pydantic import BaseModel # Stateless agent ai = LazyAgent( "anthropic", # provider string or BaseProvider instance name="analyst", system="You are a data analyst. Be precise.", model="claude-sonnet-4-6", # overrides provider default max_retries=3, # retry on 429/5xx with exponential backoff ) # Single turn, returns string answer = ai.text("Summarize Q3 results") # Single turn, returns typed Pydantic object class Report(BaseModel): summary: str findings: list[str] risk_level: str report = ai.json("Analyse Q3 results", schema=Report) # report.summary → str, report.findings → list[str] # Expose as a tool for an orchestrator analyst_tool = ai.as_tool( description="Analyse structured data and return a full report", output_schema=Report, )
Canonical last-value accessor. Returns a typed Pydantic object if output_schema was active, plain string otherwise. Always use this in pipeline code.
Full provider response. Use for metadata: .usage.input_tokens, .grounding_sources, .thinking. Semi-internal — not yet stable API.
Pass a Memory object to accumulate conversation history across turns. The object stores the full message list and replays it on every call.
from lazybridge import LazyAgent from lazybridge.core.types import Memory mem = Memory() ai = LazyAgent("anthropic") # Turn 1 ai.chat("My name is Marco.", memory=mem) # Turn 2 — history is replayed automatically ai.chat("What's my name?", memory=mem) # → "Your name is Marco." print(mem.history) # list[dict] — all user + assistant messages mem.clear() # reset for a new session # Restore from a previous session mem2 = Memory.from_history(mem.history)
stream=True returns an iterator of StreamChunk objects — each has a .delta string and a final chunk with .usage and .parsed. Every method has an async counterpart: achat, aloop, atext, ajson.
# Streaming — token by token for chunk in ai.chat("Explain transformers.", stream=True): print(chunk.delta, end="", flush=True) if chunk.is_final: print(f"\n[{chunk.usage.output_tokens} tokens]") # Async — full async/await API result = await ai.aloop("Research AI news", tools=[search_tool]) text = await ai.atext("Summarise this") obj = await ai.ajson("Extract entities", MySchema)
Pass thinking=True or a ThinkingConfig to enable chain-of-thought reasoning. The model reasons internally before answering. Access the trace via resp.thinking.
from lazybridge.core.types import ThinkingConfig resp = ai.chat( "Design the optimal architecture for a 3-agent pipeline.", thinking=ThinkingConfig(effort="high"), # "low"|"medium"|"high"|"xhigh" ) print(resp.thinking) # the reasoning trace print(resp.content) # the final answer
The single unit that makes composition possible. Wraps a Python callable or a LazyAgent. Always has a name, description, and JSON Schema. The schema is built lazily and cached.
from lazybridge import LazyTool, LazyAgent from typing import Annotated # 1. Wrap a function — schema from type hints + docstring def get_price( ticker: Annotated[str, "Stock ticker symbol e.g. AAPL"], period: str = "1y" ) -> dict: """Return OHLCV data for the given ticker.""" ... price_tool = LazyTool.from_function(get_price) # name="get_price", description="Return OHLCV data..." # schema: {"ticker": {"type":"string","description":"Stock ticker..."}, "period": ...} # 2. Wrap an agent — schema is always {"task": str} analyst = LazyAgent("anthropic", name="analyst") analyst_tool = LazyTool.from_agent(analyst, description="Analyse data") # 3. Specialize — reuse with overrides eu_tool = price_tool.specialize( name="get_eu_price", guidance="Always use European market tickers.", ) # Persist and reload across processes price_tool.save("tools/get_price.py") price_tool = LazyTool.load("tools/get_price.py")
Serialises any function-backed or agent-backed tool to a human-readable .py file. The file includes the full source, imports, and LazyTool.from_function() call. Load it in any process, machine, or project — no import of the original module needed.
Returns a copy of the tool with selective overrides — new name, description, or guidance. The schema is rebuilt; the callable is shared. Use to create role-specific variants without duplicating the function.
Holds a shared store, event log, and graph topology. The canonical way to compose multiple agents into a pipeline — as a parallel fan-out, a sequential chain, or a nested combination of both.
from lazybridge import LazyAgent, LazySession sess = LazySession(db="run.db", tracking="basic", console=True) # Parallel: all agents receive the same task, outputs concatenated parallel_tool = sess.as_tool( "research", "Research across three domains", mode="parallel", participants=[ LazyAgent("anthropic", name="tech", session=sess), LazyAgent("openai", name="market", session=sess), LazyAgent("anthropic", name="legal", session=sess), ], ) # Chain: output of each step becomes input of the next chain_tool = sess.as_tool( "pipeline", "Research → write → edit", mode="chain", participants=[parallel_tool, writer, editor], # LazyTool can be a participant )
System prompt injection evaluated at call time, not construction time. Declare the wiring at startup; evaluation happens when the agent actually runs. Safe to create before an upstream agent has executed.
from lazybridge import LazyAgent, LazyContext, LazyStore store = LazyStore() store.write("style", "Formal tone, max 300 words, cite sources.") researcher = LazyAgent("anthropic", name="researcher") fact_checker = LazyAgent("openai", name="fact_checker", context=LazyContext.from_agent(researcher)) # Composed context: researcher output + fact_checker output + store value writer = LazyAgent("anthropic", name="writer", context=( LazyContext.from_agent(researcher, prefix="[Research]") + LazyContext.from_agent(fact_checker, prefix="[Fact-check]") + LazyContext.from_store(store, keys=["style"]) )) # Execute in order — all contexts are evaluated lazily at call time researcher.loop("Fusion energy research 2025") fact_checker.chat("Verify the claims.") result = writer.chat("Write the briefing.")
A shared key-value blackboard. In-memory or SQLite-backed. The right tool when agents are intentionally decoupled — running at different times, in separate processes, or when some steps may be skipped.
mode="chain")from lazybridge import LazyStore store = LazyStore(db="pipeline.db") # SQLite, WAL mode # Sync store.write("result", agent.result, agent_id=agent.id) value = store.read("result") # Async — use inside asyncio.gather / sess.gather() await store.awrite("result", data) value = await store.aread("result") # Inject into next agent via LazyContext ctx = LazyContext.from_store(store, keys=["result"])
loop() is the tool execution engine. Pass tools, call once,
let the model drive. No manual dispatch. No response parsing. No iteration logic.
The loop terminates when the model stops requesting tools, or when max_steps is reached.
from lazybridge import LazyAgent, LazyTool drafter = LazyAgent("anthropic", system="You are a precise technical writer.") # verify= runs a quality gate after each draft. # PASS → return. FAIL → retry with feedback. Max 3 attempts. result = drafter.loop( "Write a 200-word intro to transformer architecture.", tools=[search_tool], verify="Is this accurate, clearly written, and under 200 words? " "Reply PASS or FAIL with a one-sentence reason.", max_verify=3, on_event=lambda e, p: print(e, p), ) # verify= is the canonical quality gate. Prefer it over a separate reviewer agent.
output_schema= enforces typed output at any level of the stack.
Pass a Pydantic model or a JSON Schema dict. The framework handles
native provider APIs and system prompt injection for belt-and-suspenders enforcement.
from pydantic import BaseModel from lazybridge import LazyAgent, LazySession class RiskProfile(BaseModel): rating: str # "LOW" | "MEDIUM" | "HIGH" factors: list[str] recommendation: str # On a single agent analyst = LazyAgent("anthropic", output_schema=RiskProfile) analyst.chat("Analyse the EV market") profile = analyst.result # RiskProfile instance # On the last step of a chain — type propagates up sess = LazySession() pipeline = sess.as_tool("analysis", "...", mode="chain", participants=[ LazyAgent("anthropic", name="researcher", session=sess), LazyAgent("openai", name="analyst", session=sess, output_schema=RiskProfile), # last step is typed ] ) profile = pipeline.run({"task": "Analyse EV market"}) # RiskProfile
There are three mechanisms. They are not interchangeable. Using the wrong one is the most common source of confusion.
Agents communicate by returning values. An orchestrator calls a sub-agent as a tool; the return value becomes a tool result in the conversation. This is the primary mechanism.
Agent B reads agent A's last output via the system prompt. Declared at construction, evaluated at call time. Use when agents must remain independently callable, or the topology has multiple predecessors.
A shared blackboard. Use only when agents run at different times, in separate processes, or when any step may be skipped. Not for tightly coupled sequential pipelines.
Do not use the store to pass data between tightly coupled sequential agents. Use sess.as_tool(mode="chain") instead. The store is a blackboard, not a message queue.
LazyBridge has an explicit preference order. The canonical patterns handle most real-world systems. Escape hatches exist — use them only when the canonical layer doesn't fit.
agent.as_tool(), sess.as_tool(mode="parallel"), sess.as_tool(mode="chain"), loop(verify=...). No plumbing, no asyncio, no manual context wiring.
LazyContext.from_agent(). Pull-based wiring declared at construction. Use when agents must stay independently callable or have multiple predecessors.
LazyStore + LazyContext.from_store(). For cross-process agents, persistent state, or conditionally-skipped steps.
sess.gather() (need per-agent CompletionResponse), LazyRouter (multi-destination routing, not pass/fail), manual loops (fully dynamic graph).
Every session has an event log. In-memory by default.
Add db="pipeline.db" for SQLite persistence.
Add console=True or verbose=True for real-time stdout.
from lazybridge import LazySession, Event # Development: real-time console output sess = LazySession(db="dev.db", tracking="basic", console=True) # Production: silent SQLite, query post-hoc sess = LazySession(db="prod.db", tracking="basic") # Query by event type tool_calls = sess.events.get(event_type=Event.TOOL_CALL) responses = sess.events.get(event_type=Event.MODEL_RESPONSE) # Debug system prompt assembly (verbose only) prompts = LazySession(tracking="verbose") # sess.events.get(event_type=Event.SYSTEM_CONTEXT)
LazyBridge supports four providers with a stable string alias.
The entire API — chat(), loop(), json(),
structured output, streaming — is identical across all of them.
Mix providers inside the same pipeline freely.
# Identical API across all providers for provider in ["anthropic", "openai", "google", "deepseek"]: ai = LazyAgent(provider) print(ai.text("What is 2 + 2?")) # Custom provider: implement BaseProvider from lazybridge.core.providers.base import BaseProvider class OllamaProvider(BaseProvider): default_model = "llama3" def complete(self, request) -> CompletionResponse: ... def stream(self, request) -> Iterator[StreamChunk]: ... async def acomplete(self, request) -> CompletionResponse: ... async def astream(self, request) -> AsyncIterator[StreamChunk]: ... agent = LazyAgent(OllamaProvider(model="llama3"))