# Project export: Hive AI Agents This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small. ## Project metadata - Hackathon: UC Berkeley AI Hackathon 2025 - Tagline: In just a few clicks, Hive AI Agents transforms plain-English directives into fully tested, documented, containerized micro-services. - Devpost: https://devpost.com/software/hive-ai-agents-qeh2ky - GitHub: https://github.com/ShayManor/Calhacks-2025-Backend - Demo: http://hiveaiagents.com/ - Video: https://www.youtube.com/embed/0SFfrTTqKIM?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 1 GitHub contributor(s) — shay (6 commits) ## Devpost submission (written by the team) ### Inspiration aIn just a few clicks, Hive AI Agents transforms plain-English directives into fully tested, documented, containerized micro-services and then dynamically stitches them into complex, multi-step workflows under a root orchestrator. Built on Flask, Docker, Google Cloud Run, and powered by Gemini and Anthropic LLMs, it learns which peers to include via a custom LLM-guided CSV slicing and self-optimizes through recursive agent creation—no manual wiring, no boilerplate. Inspiration We saw a growing trend in treating AI “agents” as first-class microservices that can be composed into larger workflows—borrowing best practices from event-driven microservices architecture and EDA patterns to scale agents reliably. At the same time, leaders like Meta predict that AI will function as a mid-level engineer by 2025, writing and reviewing code in real time. We wanted to build a meta-AI: an AI that not only writes code but manages, tests, deploys, and orchestrates its own creations. ### What it does Automated Agent Generation: Users post a JSON prompt (“Create an agent that summarizes research papers…”), and the backend spins up a new Flask microservice, complete with Pydantic validation, pytest suites, documentation, Docker packaging, and a Cloud Run deployment—all via LLMs. Hierarchical Orchestration: A top-level “ResearchMaster” agent exposes /research-master, accepts a pipeline of registered agents, calls each in sequence, and returns a full execution trace or error details. Service Registry & Semantic Discovery: Every agent’s metadata is appended to agents.csv and a REST /registry endpoint—allowing GPT to pick which peers to include contextually, thus avoiding prompt bloat and ensuring relevance. ### How we built it We chose Flask for its minimal footprint and ease of writing microservices. Each generated agent lives in its own folder, and we use os.makedirs(..., exist_ok=True) to avoid mkdir collisions. pipreqs (with --mode no-pin) auto-generates requirements.txt without strict version pins—preventing deployment breakage on numpy upgrades. For orchestration we integrated Orkes Conductor via the Python SDK, defining three tasks (choose_or_create_agent, ask_agent, log_usage) and a simple DAG—all running alongside Flask in the same container. Containers deploy to Google Cloud Run with a --revision-suffix to enforce unique revisions on every build. ### Challenges we ran into Name collisions & idempotency: Re-deploying the same agent name caused Cloud Run ALREADY_EXISTS errors; fixed by adding a random suffix per deploy. Prompt context bloat: Feeding the entire registry to GPT exceeded token limits; solved with a GPT-guided CSV slicer that asks an assistant which agents matter most, then clamps to 30 rows. Silent failures: Initial 500s from uncaught exceptions left us blind; we added a global sys.excepthook to log full stack traces and wrapped each expensive step (build_agent, gcloud, ping_gpt) in try/except blocks for clear JSON errors. ### Accomplishments we're proud of Recursive agent creators: We built agents that generate agents—e.g. documentation generators, test writers, and even a new agent factory—demonstrating true self-improvement. Seamless orchestration: The three-node Conductor workflow immediately visualizes each request’s path in the Orkes UI, complete with retries, SLA timeouts, and audit logs. Plug-and-play registry: Judges can hit /registry in a browser, see every agent spun up, and chain them in new pipelines on the fly. ### What we learned Microservices patterns like the Service Registry are invaluable for AI workflows—treat agents as replaceable, discoverable services rather than monolithic code. Prompt engineering is just as critical in system architecture: crafting clear system prompts and fallback logic can make or break reliability under token constraints. DevOps for AI requires new guardrails (constraints files, revision suffixes, file locks) to handle the unpredictability of LLM-generated code. ### What's next We plan to add parallel fan-out, where multiple agent variants run concurrently and the platform selects the best output. We’ll explore dynamic sub-workflows that fork mid-pipeline based on runtime signals. Finally, we aim to launch a dashboard UI to visualize agent trees, live executions, and semantic-search-powered recommendations—all in real time. ## README (from the GitHub repository) # Calhacks-2025-Backend ## Overview **Calhacks-2025-Backend** is an advanced AI-driven platform designed to create, manage, and orchestrate AI agents—automatically. It is an "AI agent to create AI agents," enabling users, researchers, and developers to automate complex workflows by dynamically generating, deploying, and coordinating specialized micro-agents for a wide range of tasks. ## What Does This Project Do? ### 1. **Automated Agent Creation** - **Natural Language to Agent:** Users can describe the functionality they want in plain English, and the system will generate a new AI agent (microservice) tailored to that description. - **End-to-End Pipeline:** The platform handles everything from code generation, testing, and documentation to containerization and cloud deployment—without manual intervention. ### 2. **Hierarchical Agent Orchestration** - **Root Orchestrator:** At the top level, a "ResearchMaster" agent coordinates a hierarchy of sub-agents, each responsible for a specific task in a larger workflow (e.g., scholarly research, data analysis, report generation). - **Dynamic Composition:** Agents can be composed, nested, and chained together, allowing for the creation of complex, multi-step pipelines. ### 3. **Specialized Micro-Agents** - **Task-Specific Agents:** The system can generate agents for a wide variety of tasks, such as: - Crawling and parsing research papers from online databases - Extracting and deduplicating citations - Downloading and processing PDFs - Running machine learning models and statistical analyses - Generating summaries, visualizations, and reports - Handling notifications, scheduling, and resource monitoring - **Plug-and-Play:** Each agent is a self-contained microservice with its own API, documentation, and health checks. ### 4. **Agent Registry and Discovery** - **Centralized Registry:** All created agents are registered with metadata, documentation, and endpoints, making them discoverable and reusable for future workflows. - **Semantic Search:** Users can search for existing agents by describing their needs, and the system will recommend the most relevant agents based on semantic similarity. ### 5. **Automated Documentation and Testing** - **Auto-Generated Docs:** Every agent comes with automatically generated documentation, including usage examples and API references. - **Continuous Testing:** Agents are tested automatically before deployment to ensure reliability and correctness. ### 6. **Cloud-Native Deployment** - **Containerization:** Each agent is packaged as a Docker container. - **One-Click Cloud Deploy:** Agents are deployed to the cloud (e.g., Google Cloud Run) with public endpoints, ready to be integrated into larger systems. ### 7. **Extensible and Modular** - **Custom Pipelines:** Users can build custom pipelines by chaining together existing agents or creating new ones on demand. - **Integration Ready:** Agents can interact with external APIs, databases, and other services as needed. ### 8. **Example Use Case: Scholarly Research Pipeline** The platform can automatically build an end-to-end research assistant that: - Crawls academic databases for papers - Extracts and cleans metadata, citations, and content - Runs topic modeling and trend analysis - Generates executive summaries and visualizations - Notifies users of new insights via email or Slack ### 9. **Interactive Q&A and Visualization** - **Question Answering:** Specialized agents can answer user questions about the research dataset, fetch relevant document snippets, and maintain conversational context. - **Data Visualization:** Agents can generate charts, graphs, and dashboards from analysis results. ### 10. **Security, Monitoring, and Maintenance** - **Access Control:** Agents can enforce API key and role-based access controls. - **Resource Monitoring:** The system tracks resource usage and health of all agents, with automated alerts and restarts on failure. - **Automated Cleanup:** Old logs, temporary files, and unused resources are pruned automatically. --- ## Key Features at a Glance - **AI that builds AI:** Describe what you want, and the system creates, tests, documents, and deploys a new agent for you. - **Hierarchical orchestration:** Build complex workflows by composing specialized agents. - **Registry and search:** Discover and reuse agents with semantic search. - **Cloud-native:** Agents are containerized and deployed with public endpoints. - **Automated documentation and testing:** Every agent is production-ready out of the box. - **Extensible:** Easily add new capabilities or integrate with external systems. --- ## Who Is This For? - **Researchers:** Automate literature reviews, data extraction, and analysis. - **Developers:** Rapidly prototype and deploy new AI-powered microservices. - **Organizations:** Build scalable, maintainable AI workflows with minimal manual effort. --- ## Example: Creating a New Agent 1. **Describe your agent:** _"Create an agent that summarizes research papers and emails the summary to my team."_ 2. **The system will:** - Generate the code for the agent - Test and document it - Deploy it to the cloud - Register it for future use 3. **Result:** You get a ready-to-use API endpoint for your new agent, complete with documentation and integration options. --- ## Why Is This Unique? - **Meta-AI:** This is not just an AI agent—it is an AI that creates, manages, and orchestrates other AI agents, enabling a new level of automation and scalability. - **No manual coding required:** Go from idea to deployed microservice in minutes, using only natural language. - **Self-improving:** The system can generate agents to improve itself, such as documentation generators, test writers, and even new agent creators. This recursive capability means the platform can evolve and expand its own ecosystem of agents over time, adapting to new requirements and domains with minimal human intervention. --- ## Summary **Calhacks-2025-Backend** is a groundbreaking platform that automates the creation, deployment, and orchestration of AI agents. By leveraging natural language, users can rapidly build complex, production-ready AI workflows—without writing code or managing infrastructure. The system's modular, extensible, and self-improving architecture makes it ideal for research, development, and enterprise automation. --- ## Getting Started > _For installation, setup, and API usage instructions, please see the [Usage Guide](#) (to be completed)._ --- ## License MIT License (or specify your license here) --- ## Contact For questions, feature requests, or contributions, please open an issue or contact the maintainers. --- **Calhacks-2025-Backend**: AI that builds AI—so you can focus on what matters most. ## Detected evidence (automated analysis) Indexed codebase: 11 recognized source files, 48 KB. - Python (language) — detected in the code - Flask (technology) — claimed on Devpost, not found in the code - Google Gemini (technology) — claimed on Devpost, not found in the code ## Codebase structure (from repository index) ### Files (14 of 14) ``` .gitignore app.py create_on_letta.py examples/app_example.py examples/Dockerfile_example examples/fetch_wrapper_example.py main.py README.md registry.py services/agents.csv tests/test_agents.py tests/test_fetch_mesh.py tests/test_gmaps_api.py tests/test_letta_router.py ``` ### Dependencies No dependency index available. ### Recent commits (newest first) - main - FUCKSHITBASICALLYOWKRSUFCK - Works but errors out - Works without pinging other agent - Build the actual agents (50%) - Agents upload, GCR works Letta 50/50 and Fetch No - Initial commit ## Key source files (fetched from GitHub, selected and truncated for size) ### app.py ```python import json import logging import os import sys import time import traceback import uuid import openai from dotenv import load_dotenv from flask import Flask, jsonify, request from letta_client import Letta from registry import AgentRegistry from services.create_agent import create_agent from pathlib import Path, PurePosixPath def log_exceptions(exc_type, exc, tb): logging.error("UNCAUGHT EXCEPTION:\n%s", "".join(traceback.format_exception(exc_type, exc, tb))) sys.excepthook = log_exceptions app = Flask(__name__) AGENTS_CSV = Path(__file__).resolve().parent / "agents.csv" @app.route("/ping") def ping(): return jsonify({'ping': 'pong'}) load_dotenv() reg, letta = AgentRegistry(), Letta(token=os.environ["LETTA_API_KEY"]) @app.post("/prompt") def handle(): print("test") # data = request.get_json(force=True) user_prompt = "Find me 10 things to do in berkeley California this saturday for someone who is 21." agent = reg.search(user_prompt) if agent is None: # cold start name = f"A{uuid.uuid4().hex[:6]}" letta_id = create_agent(user_prompt, None) reg.add(name, user_prompt, letta_id) agent = reg.search(user_prompt) # guaranteed hit # short-running → sync call to Letta resp = letta.agents.messages.create( agent_id=agent.id, messages=[{"role": "user", "content": user_prompt}], ) return jsonify({"agent": agent.name, "response": resp.messages[-1].content}) GMAPS = """ You are to rebuild the Google Maps Utility Agent as a Flask microservice using the Google Maps Platform (Geocoding API + Places API v1). Follow these exact specifications: 1 Overall Service Requirements Expose four JSON POST endpoints: /geocode, /reverse_geocode, /nearby, /textsearch. Use Flask for routing, Pydantic for request validation, and requests (or httpx) for outbound calls. Use this GOOGLE_API_KEY everywhere:AIzaSyARrvLp5k2yA23PoawhD6bcpND3qcA3boA Return HTTP 200 with valid JSON on success; map Google’s "ZERO_RESULTS" → empty list or 404 where appropriate; propagate upstream errors as 502 or 400. 2 Endpoint Specifications 2.1 /geocode Input: json Copy Edit { "address": "1600 Amphitheatre Pkwy, Mountain View CA" } Validation: Pydantic AddressIn(address: str). developers.google.com Call: pgsql Copy Edit GET https://maps.googleapis.com/maps/api/geocode/json ?address={address}&key={API_KEY} per Geocoding API docs developers.google.com . Success: { "lat": , "lng": , "formatted_address": "", "place_id": "" } Errors: 404 if "results" is empty. 502 for non-200 HTTP. 400 if Google returns status ≠ OK|ZERO_RESULTS. 2.2 /reverse_geocode Input: { "lat": 37.422, "lng": -122.084 } Validation: Pydantic LatLngIn(lat: float, lng: float). Call: GET https://maps.googleapis.com/maps/api/geocode/json ?latlng={lat},{lng}&key={API_KEY} per reverse geocoding in the same Geocoding API. Success: { "formatted_address": "", "place_id": "" } Errors: 404 if no results. Same HTTP/status mapping as /geocode. 2.3 /nearby (Places API – New) Input: { "lat": 37.422, "lng": -122.084, "type": "restaurant", "radius": 500 } Validate with NearbyIn(lat: float, lng: float, type: str, radius: int) and 1 ≤ radius ≤ 50 000. Call: POST https://places.googleapis.com/v1/places:searchNearby Headers: X-Goog-Api-Key: {API_KEY} X-Goog-FieldMask: places.displayName,places.formattedAddress JSON body: { "locationRestriction": { "circle": { "center": { "latitude": lat, "longitude": lng }, "radius": radius } }, "includedTypes": [ type ] } per Nearby Search (New) docs. Success: { "places": [ /* Place objects with displayName & formattedAddress */ ] } Errors: 400 if missing required field mask. 502 for HTTP ≠ 200. 2.4 /textsearch (Places API – New) Input: { "query": "pizza in New York", "lat": , "lng": , "radius": } Validate with TextSearchIn(query: str, lat: float | None, lng: float | None, radius: int | None). Call: POST https://places.googleapis.com/v1/places:searchText Headers: X-Goog-Api-Key: {API_KEY} X-Goog-FieldMask: places.displayName,places.formattedAddress JSON body: { "textQuery": query, // optional: "locationRestriction": { "circle": { "center": { "latitude": lat, "longitude": lng }, "radius": radius } } } per Text Search (New) reference Success: { "results": [ /* Place objects */ ] } 3 Error Handling & Best Practices Always check data["status"] against {"OK","ZERO_RESULTS"} before using data["results"]. Timeouts: use timeout=10 seconds on all HTTP calls. Logging: on non-200 or status≠OK, log the raw response and return structured JSON { "error": ... }. FieldMask: required on all new Places v1 calls or you’ll get HTTP 400. developers.google.com 4 Flask + Pydantic Boilerplate Example from flask import Flask, request, jsonify from pydantic import BaseModel, Field, ValidationError import os, requests API_KEY = os.environ["GOOGLE_API_KEY"] GEOCODE_URL = "https://maps.googleapis.com/maps/api/geocode/json" NEARBY_URL = "https://places.googleapis.com/v1/places:searchNearby" TEXT_URL = "https://places.googleapis.com/v1/places:searchText" app = Flask(__name__) class AddressIn(BaseModel): address: str = Field(...) # (Define LatLngIn, NearbyIn, TextSearchIn similarly) def gmap_get(url, params): params["key"] = API_KEY r = requests.get(url, params=params, timeout=10) # ...status & JSON checks as above... """ @app.route("/create_agent", methods=["POST"]) def create(): start = time.time() res = """ Create an agent that has 5 endpoints. Each endpoint should either ping chatGPT or claude. Make sure the name has something to do with text as these models only do text. The option for models should be gpt4.1 nano, gpt4.1, gpt o3, claude haiku 3.5, claude sonnet 4. It should work well and be easy to use with relevant and good docs and description. It sh [truncated — 5003 more characters] ``` ### main.py ```python import requests, time, json, logging logging.basicConfig(level=logging.INFO) def bulk_create(prompts: list[str], base_url="http://localhost:5000"): """ For each prompt in `prompts`, attempt to create an agent by calling /create_agent up to two times. If the server rejects POST with 405, fall back to GET?prompt=<...>, retry once on failure, then skip. Returns a dict mapping each prompt to the JSON response or None. """ results: dict[str, dict | None] = {} for prompt in prompts: for attempt in (1, 2): try: # Try POST first url = f"{base_url}/create_agent" logging.info("Creating agent for prompt (attempt %d): %s", attempt, prompt[:50]) resp = requests.post( url, json={"prompt": prompt}, timeout=420 ) if resp.status_code == 405: # METHOD NOT ALLOWED: fall back to GET logging.warning("POST returned 405, retrying with GET") resp = requests.post( url, params={"prompt": prompt}, timeout=420 ) resp.raise_for_status() # HTTPError for 4xx/5xx # parse JSON data = resp.json() results[prompt] = data logging.info("✓ Success for prompt: %s (%.2fs)", prompt[:40], resp.elapsed.total_seconds()) break # stop retrying on success except requests.HTTPError as he: logging.warning("HTTP error on attempt %d for '%s': %s", attempt, prompt[:40], he) except requests.RequestException as re: logging.warning("Request exception on attempt %d for '%s': %s", attempt, prompt[:40], re) except json.JSONDecodeError as je: logging.warning("Invalid JSON response on attempt %d for '%s': %s", attempt, prompt[:40], je) except Exception as e: logging.warning("Unexpected error on attempt %d for '%s': %s", attempt, prompt[:40], e) # on failure, record None if last attempt if attempt == 2: results[prompt] = None time.sleep(1) # brief pause before retry time.sleep(10) return results if __name__ == '__main__': prompts = [ "Define the root agent “ResearchMaster” with the responsibility of orchestrating all sub-agents in a scholarly research pipeline.", "Under “ResearchMaster”, create a “Crawler” agent that collects research papers from multiple online scholarly databases.", "Under “Crawler”, create a “LinkParser” agent that parses each paper’s reference links and removes duplicates.", "Under “Crawler”, create a “ContentFetcher” agent that downloads HTML or PDF content for every parsed link.", "Under “ContentFetcher”, create a “CleanHTML” agent that sanitizes HTML and strips out scripts, ads, and navigation.", "Under “CleanHTML”, create a “TextExtractor” agent that extracts plain text from sanitized HTML fragments.", "Under “TextExtractor”, create a “LanguageDetector” agent that filters out non-English documents.", "Under “LanguageDetector”, create an “EnglishFilter” agent that retains only documents detected as English.", "Under “EnglishFilter”, create a “MetadataExtractor” agent that pulls titles, authors, dates, and journal info.", "Under “MetadataExtractor”, create a “CitationExtractor” agent that identifies in-text citations and bibliography entries.", "Under “CitationExtractor”, create a “Deduplicator” agent that merges multiple references to the same work.", "Under “Deduplicator”, create a “PDFDownloader” agent that retrieves PDF files for each unique citation.", "Under “PDFDownloader”, create a “PDFTextExtractor” agent that uses OCR or PDF parsing to extract raw text.", "Under “PDFTextExtractor”, create a “SectionSplitter” agent that breaks papers into Introduction, Methods, Results, etc.", "Under “SectionSplitter”, create an “AbstractExtractor” agent that isolates and stores each paper’s abstract.", "Under “AbstractExtractor”, create a “KeywordExtractor” agent that identifies the top 10 keywords per paper.", "Under “KeywordExtractor”, create a “TopicModeler” agent that clusters papers into thematic groups via LDA.", "Under “TopicModeler”, create a “ClusterLabeler” agent that assigns human-readable labels to each topic cluster.", "Under “ClusterLabeler”, create a “TrendDetector” agent that analyses publication dates to surface emerging topics.", "Under “TrendDetector”, create a “SentimentAnalysis” agent that gauges the tone of abstracts over time.", "Under “SentimentAnalysis”, create a “SummaryGenerator” agent that composes concise executive summaries of sentiment trends.", "Under “SummaryGenerator”, create an “ExecutiveBrief” agent that formats summaries into slide-ready bullet points.", "Under “ExecutiveBrief”, create a “ChartGenerator” agent that uses Matplotlib to draw trend charts.", "Under “ChartGenerator”, create a “ReportAssembler” agent that combines charts and text into a PDF report.", "Under “ReportAssembler”, create a “FormatMapper” agent that converts the PDF into HTML and Markdown versions.", "Under “FormatMapper”, create a “PDFReporter” agent that emails the final report to stakeholders.", "Under “PDFReporter”, create an “EmailNotifier” agent that sends success/failure notifications to a Slack channel.", "Under “ResearchMaster”, create an “Analyzer” agent that runs statistical and machine-learning models [truncated — 8379 more characters] ``` ### create_on_letta.py ```python import os import dotenv from letta_client import Letta def upload_to_letta(name: str, prompt: str): dotenv.load_dotenv() key = os.getenv("LETTA_API_KEY") id = 'cf1f587f-461d-44c7-8511-4b7cdd7839f8' client = Letta(token=key) agent_state = client.agents.create( name=name, system=prompt, llm_config={ "model": "openai/gpt-4o-mini", "model_endpoint_type": "openai", "context_window": 8192, "temperature": 0.2, }, ) print("New agent id:", agent_state.id) return agent_state.id ``` ### registry.py ```python # registry.py from dataclasses import dataclass, field from sentence_transformers import SentenceTransformer from pathlib import Path import numpy as np, uuid, pickle ENCODER = SentenceTransformer("all-MiniLM-L6-v2") REGISTRY_FILE = Path("agents.pkl") @dataclass class AgentMeta: id: str name: str sys_prompt: str embedding: np.ndarray = field(repr=False) class AgentRegistry: def __init__(self): self.agents: list[AgentMeta] = [] if REGISTRY_FILE.exists() and REGISTRY_FILE.stat().st_size: self.agents = pickle.loads(REGISTRY_FILE.read_bytes()) def _save(self): REGISTRY_FILE.write_bytes(pickle.dumps(self.agents)) def search(self, prompt: str, thresh: float = 0.83) -> AgentMeta | None: if not self.agents: return None q = ENCODER.encode(prompt, normalize_embeddings=True) sims = [float(q @ a.embedding) for a in self.agents] best = max(range(len(sims)), key=sims.__getitem__) return self.agents[best] if sims[best] > thresh else None def add(self, name: str, sys_prompt: str, letta_id: str): vec = ENCODER.encode(sys_prompt, normalize_embeddings=True) self.agents.append(AgentMeta(letta_id, name, sys_prompt, vec)) self._save() ``` ### examples/app_example.py ```python import os from flask import Flask, request, jsonify app = Flask(__name__) @app.route('/', methods=["GET"]) def get(): return jsonify({'test': '123'}) @app.route('/ping', methods=["GET"]) def ping(): return jsonify({'ping': 'pong'}) # Your code here if __name__ == "__main__": port = int(os.environ.get("PORT", 8080)) app.run(host="0.0.0.0", port=port) ``` ### tests/test_gmaps_api.py ```python import os, pytest, httpx, math, time BASE = "https://google-geocode-agent-854967522738.us-central1.run.app" def close(a, b, eps=1e-3): return abs(a - b) < eps @pytest.mark.parametrize("address", [ "1600 Amphitheatre Pkwy, Mountain View CA", "1 Market St, San Francisco CA" ]) def test_geocode_reverse(address): with httpx.Client(timeout=10.0) as client: g = client.post(f"{BASE}/geocode", json={"address": address}) assert g.status_code == 200, g.text data = g.json() lat, lng = data["lat"], data["lng"] r = client.post(f"{BASE}/reverse_geocode", json={"lat": lat, "lng": lng}) assert r.status_code == 200, r.text rev = r.json() assert address.split(",")[0].lower() in rev["formatted_address"].lower() def test_nearby_result_shape(): here = {"lat": 37.422, "lng": -122.084} body = {**here, "type": "restaurant", "radius": 500} with httpx.Client() as client: res = client.post(f"{BASE}/nearby", json=body) assert res.status_code == 200, res.text places = res.json()["places"] assert isinstance(places, list) @pytest.mark.skipif("GOOGLE_API_KEY" not in os.environ, reason="Requires quota to hit live Google Text Search") def test_textsearch_live(): with httpx.Client() as client: res = client.post(f"{BASE}/textsearch", json={"query": "pizza in Palo Alto"}) assert res.status_code == 200 assert "results" in res.json() ``` ### tests/test_fetch_mesh.py ```python import asyncio, os, threading, time, httpx, json import pytest, respx from uagents import Agent, Model, Context from examples.fetch_wrapper_example import service as wrapper, _call_letta @pytest.fixture(scope="session", autouse=True) def start_wrapper(): # spin up the uAgent in a thread so tests can hit :8000 t = threading.Thread(target=wrapper.run, daemon=True) t.start() time.sleep(2) # give it a moment yield # wrapper.run exits with the program so nothing to clean up class Prompt(Model): prompt: str @respx.mock @pytest.mark.asyncio async def test_rest_prompt(): # mock Letta respx.post("https://api.letta.com/v1/agents/").mock( return_value=httpx.Response(200, json={"messages": [{"content": "pong"}]})) async with httpx.AsyncClient() as client: r = await client.post("http://localhost:8000/prompt", json={"prompt": "ping"}) assert r.status_code == 200 body = r.json() assert body["result"] == "pong" assert "agent_id" in body @respx.mock @pytest.mark.asyncio async def test_p2p_loopback(): # Arrange fake letta again respx.post("https://api.letta.com/v1/agents/").mock( return_value=httpx.Response(200, json={ "messages": [{"content": "mesh ok"}]})) # Use uAgents in-memory deliverer loop = asyncio.get_running_loop() fut = loop.create_future() @wrapper.on_message(model=Prompt) async def _tmp(ctx: Context, sender: str, msg: Prompt): fut.set_result(msg.prompt) await wrapper.send(wrapper.address, Prompt(prompt="mesh ok")) assert await asyncio.wait_for(fut, 3) == "mesh ok" ``` ### tests/test_letta_router.py ```python import os, json, socket, subprocess, time, tempfile, uuid, importlib.util from pathlib import Path import httpx, pytest, respx from fastapi.encoders import jsonable_encoder from sentence_transformers import SentenceTransformer from registry import AgentRegistry # ← your module from services.create_agent import create_agent # real func, mocked below LETTA_URL = "https://api.letta.com/v1" @pytest.fixture(scope="session") def router(tmp_path_factory): """Fresh registry file in a temp dir so we don't clobber prod.""" tmp_dir = tmp_path_factory.mktemp("router") (tmp_dir / "agents.pkl").write_bytes(b"") # empty registry # monkey-patch expected global path import registry as reg_mod; reg_mod.REGISTRY_FILE = tmp_dir/"agents.pkl" return AgentRegistry() def test_router_reuse(router, monkeypatch): enc = SentenceTransformer("all-MiniLM-L6-v2") prompt = "Write me a haiku about quantum entanglement" # fake first agent creation monkeypatch.setattr("services.create_agent.create_agent", lambda name, p=prompt: f"let-{uuid.uuid4()}") agent = router.search(prompt) if agent is None: # cold-start router.add("A0", prompt, "let-123") # second call must find same agent assert router.search(prompt).id == router.agents[0].id def fake_create_agent(name, p): r = httpx.post(f"{LETTA_URL}/agents/", json={"name": name, "prompt": p}) return r.json()["id"] @respx.mock def test_router_cold_start(monkeypatch, router): # 1 mock the outgoing POST route = respx.post(f"{LETTA_URL}/agents/").mock( return_value=httpx.Response(201, json={"id": "let-new"}) ) # 2 stub create_agent so it *makes* that POST def fake_create_agent(name, prompt): resp = httpx.post(f"{LETTA_URL}/agents/", json={"name": name, "prompt": prompt}) return resp.json()["id"] monkeypatch.setattr("services.create_agent.create_agent", fake_create_agent) # 3 trigger cold-start logic (router sees no match) prompt = "Kubernetes YAML won't apply, help" assert router.search(prompt) is None # still cold router.add("A1", prompt, fake_create_agent("A1", prompt)) # 4 now the mock should have been hit exactly once assert route.call_count == 1 ``` ### examples/fetch_wrapper_example.py ```python """ fetch_wrapper_example.py ------------------------ Generic adapter that exposes *any* Letta agent as both (1) a Fetch.ai uAgent endpoint (P2P + REST) and (2) a simple HTTP /prompt route for local health checks. Copy this file unchanged into every container; pass two env-vars: LETTA_AGENT_ID – the UUID of the Letta agent this wrapper fronts LETTA_API_KEY – your Letta API key Optional env-vars: UAGENT_PORT – port for the uAgent (default 8000) LETTA_BASE_URL – override if self-hosting Letta FETCH_SEED – deterministic seed for agent keys """ import os, asyncio, httpx from typing import Optional from dotenv import load_dotenv from uagents import Agent, Model, Context load_dotenv() LETTA_URL = os.getenv("LETTA_BASE_URL", "https://api.letta.com/v1") LETTA_KEY = os.environ["LETTA_API_KEY"] AGENT_ID = os.environ["LETTA_AGENT_ID"] AGENT_PORT = int(os.getenv("UAGENT_PORT", "8000")) # ---------- Data models ---------- class Prompt(Model): prompt: str class Answer(Model): result: str agent_id: str # ---------- Helper that calls Letta ---------- async def _call_letta(prompt: str) -> str: headers = {"Authorization": f"Bearer {LETTA_KEY}"} payload = {"messages": [{"role": "user", "content": prompt}]} async with httpx.AsyncClient(timeout=60) as client: r = await client.post(f"{LETTA_URL}/agents/{AGENT_ID}/messages", json=payload, headers=headers) r.raise_for_status() return r.json()["messages"][-1]["content"] # ---------- uAgent definition ---------- service = Agent( name=f"wrapper-{AGENT_ID[:6]}", port=AGENT_PORT, endpoint=[f"http://0.0.0.0:{AGENT_PORT}/submit"], seed=os.getenv("FETCH_SEED", "auto-generated"), ) @service.on_message(model=Prompt, replies=Answer) async def handle_prompt(ctx: Context, sender: str, msg: Prompt): ctx.logger.info("Received from %s: %s", sender, msg.prompt[:120]) reply = await _call_letta(msg.prompt) await ctx.send(sender, Answer(result=reply, agent_id=AGENT_ID)) # Optional REST surface (useful on Cloud Run) @service.on_rest_post("/prompt", expected_model=Prompt, responses=[Answer]) async def rest_prompt(ctx: Context, data: Prompt): reply = await _call_letta(data.prompt) return Answer(result=reply, agent_id=AGENT_ID) if __name__ == "__main__": service.run() ``` ### tests/test_agents.py ```python # test_agents.py """ End-to-end smoke-tests for the research-pipeline agents. Each test does three things: 1. Calls the agent’s *health* endpoint and asserts HTTP 200 + JSON body. 2. Sends a minimal, valid POST payload to the primary endpoint. 3. Verifies both HTTP 200 and a couple of mandatory JSON keys. The tests are intentionally lightweight (≤ 2 s total) so you can run them in CI on every deploy. """ import json import base64 import httpx import pytest # # ---------------------------- helper ---------------------------- # def post_ok(url: str, payload: dict, must_have: set[str]) -> None: """POST payload → url and assert 200 + required keys.""" with httpx.Client(timeout=15) as c: r = c.post(url, json=payload) assert r.status_code == 200, r.text data = r.json() missing = must_have - data.keys() print(must_have) assert not missing, f"response missing keys: {missing}" # # ------------------------- text-respondent ----------------------- # TR_BASE = "https://text-respondent-854967522738.us-central1.run.app" @pytest.mark.parametrize("model,endpoint", [ ("gpt4.1-nano", "/text-generate"), ("gpt4.1", "/text-summarize"), ("gpt-o3", "/text-analyze"), ("claude-haiku", "/text-rewrite"), ("claude-sonnet","/text-translate"), ]) def test_text_respondent(model, endpoint): # health with httpx.Client(timeout=10) as c: r = c.get(f"{TR_BASE}/ping") assert r.status_code == 200 # minimal happy-path payloads if endpoint == "/text-generate": post_ok(f"{TR_BASE}{endpoint}", {"prompt": "Hello world", "model": model}, {"prompt", "response"}) elif endpoint == "/text-summarize": post_ok(f"{TR_BASE}{endpoint}", {"text": "One two three four five six seven eight nine ten " * 3, "model": model}, {"original_len", "summary"}) elif endpoint == "/text-analyze": post_ok(f"{TR_BASE}{endpoint}", {"text": "Hello world", "model": model}, {"word_count", "sentiment"}) elif endpoint == "/text-rewrite": post_ok(f"{TR_BASE}{endpoint}", {"text": "Rewrite me", "style": "casual", "model": model}, {"rewritten", "style"}) else: # translate post_ok(f"{TR_BASE}{endpoint}", {"text": "Hello", "target_language": "es", "model": model}, {"translation", "target_language"}) # # ------------------------- research-master ----------------------- # RM_BASE = "https://researchmaster-854967522738.us-central1.run.app" def test_research_master_health_and_registry(): with httpx.Client(timeout=10) as c: assert c.get(f"{RM_BASE}/health").status_code == 200 # registry should be json list even if empty r = c.get(f"{RM_BASE}/agents") assert r.status_code == 200 assert isinstance(r.json(), list) def test_research_master_pipeline(): pipeline = [ {"name": "researchmaster-crawler", "url": "https://researchmaster-crawler-854967522738.us-central1.run.app"}, ] post_ok(f"{RM_BASE}/research-master", {"query": "quantum computing", "pipeline": pipeline}, {"trace", "result"}) # # ---------------------------- crawler ---------------------------- # CRAWLER_BASE = "https://researchmaster-crawler-854967522738.us-central1.run.app" def test_crawler_health_and_search(): with httpx.Client(timeout=10) as c: assert c.get(f"{CRAWLER_BASE}/crawler/health").status_code == 200 post_ok(f"{CRAWLER_BASE}/crawler/search", {"query": "graph neural networks", "max_results": 3}, {"papers"}) # # ---------- content-fetcher → clean-html → lang-detect ---------- # CF_BASE = "https://contentfetcher-854967522738.us-central1.run.app" CH_BASE = "https://cleanhtml-854967522738.us-central1.run.app" LD_BASE = "https://lang-detect-854967522738.us-central1.run.app" def test_content_fetcher_chain(): # fetch a known small HTML page (example.com) post_ok(f"{CF_BASE}/crawler/content-fetcher", {"links": ["http://example.com"], "timeout": 5}, {"results"}) # clean html html = "Hello" post_ok(f"{CH_BASE}/clean-html", {"html": html}, {"cleaned_html", "extracted_text"}) # language detect post_ok(f"{LD_BASE}/language-detect", {"text": "This is an English sentence."}, {"language", "is_english"}) # # --------------- citation extractor → deduplicator -------------- # CE_BASE = "https://citation-extractor-854967522738.us-central1.run.app" DD_BASE = "https://citation-duplicate-merger-854967522738.us-central1.run.app" def test_citation_and_dedupe(): sample = ("Smith, J. (2022). Deep Learning. Journal of AI.\n" "Smith, J. (2022). Deep Learning. Journal of AI.") post_ok(f"{CE_BASE}/citation-extractor", {"text": sample}, {"bibliography"}) citations = [ {"title": "Deep Learning", "authors": "Smith", "year": 2022}, {"title": "Deep Learning", "authors": "Smith", "year": 2022}, ] post_ok(f"{DD_BASE}/deduplicate", {"citations": citations}, {"deduplicated", "removed"}) # # -------- pdf-downloader → pdf-text-extractor → section-splitter - # PD_BASE = "https://pdf-downloader-854967522738.us-central1.run.app" PT_BASE = "https://pdf-text-extractor-854967522738.us-central1.run.app" SS_BASE = "https://section-splitter-854967522738.us-central1.run.app" def test_pdf_flow(): link = "https://arxiv.org/pdf/2107.03374.pdf" # download small PDF r = httpx.post(f"{PD_BASE}/download-pdfs", json={"citations": [{"title": "Any", "pdf_link": link}]}, timeout=30) assert r.status_code == 200 pdf_url = link # in success case we reus [truncated — 394 more characters] ```