Full Example¶
A complete working script showing the Zubbl SDK end-to-end.
Setup¶
Complete Script¶
"""
Zubbl SDK — Complete Example
Run: python example.py
"""
from zubbl import ZubblClient
# 1. Initialize
zubbl = ZubblClient(
api_key="zubbl_YOUR_TENANT_YOUR_SECRET", # Get from https://app.zubbl.tech
base_url="https://api.zubbl.tech",
)
# 2. Check usage
usage = zubbl.get_usage()
print(f"Trajectories: {usage.trajectories_ingested}")
print(f"Credits: {usage.credits_remaining}")
# 3. Query for recommendations before running a task
policy = zubbl.query(task_description="code review")
if policy:
print(f"Recommended approach: {policy.task_type}")
print(f"Confidence: {policy.confidence_score}")
else:
print("No policies yet — they build up as you ingest more trajectories")
# 4. Record a trajectory (your agent's execution)
trajectory_id = zubbl.start_trajectory(
task_description="Review main.py for code quality issues",
metadata={"project": "my-app"},
)
zubbl.record_step(
action="read_file",
tool_name="file_reader",
tool_input={"file": "main.py"},
tool_output="def hello(): print('world')",
tokens_used=100,
latency_ms=200,
)
zubbl.record_step(
action="analyze",
tool_name="ast_parser",
tool_input={"check": "code_quality"},
tool_output="No issues found",
tokens_used=250,
latency_ms=800,
)
zubbl.record_step(
action="respond",
tool_name="output",
tool_input={"format": "markdown"},
tool_output="Code looks clean. No issues detected.",
tokens_used=100,
latency_ms=300,
)
# End trajectory — auto-ingests to Zubbl for learning
trajectory = zubbl.end_trajectory(status="success")
print(f"Trajectory {trajectory_id} ingested!")
# 5. Submit feedback
zubbl.feedback(
trajectory_id=trajectory_id,
rating=0.85,
comment="Accurate review, fast execution",
)
print("Feedback submitted!")
# 6. Check updated usage
usage = zubbl.get_usage()
print(f"Trajectories: {usage.trajectories_ingested}")
print(f"Success rate: {usage.task_success_rate:.1%}")
# 7. Check failure stats
stats = zubbl.get_failure_stats()
print(f"Failures seen: {stats['total_failures_seen']}")
print(f"Auto-recovered: {stats['total_auto_recovered']}")
print("\nDone! Check your dashboard at https://app.zubbl.tech")
Wrap an Existing Agent (Even Simpler)¶
from zubbl import ZubblClient
zubbl = ZubblClient(api_key="zubbl_YOUR_KEY")
# Wrap any callable — Zubbl handles everything automatically
smart_agent = zubbl.wrap(your_existing_agent)
result = smart_agent.run("Your task here")
# That's it — trajectories are recorded, policies are applied,
# failures trigger reflexion, and your agent improves over time.
Framework Examples¶
LangChain¶
from langchain.agents import create_openai_tools_agent
from zubbl import ZubblClient
zubbl = ZubblClient(api_key="zubbl_YOUR_KEY")
agent = create_openai_tools_agent(llm, tools, prompt)
smart_agent = zubbl.wrap(agent)
result = smart_agent.invoke({"input": "Analyze this codebase"})