e6685203fe
- Added REPO_INVENTORY.md with all repos, branches, remotes, and staging info - Added .gitignore - Synced all existing docs from local workspace - Centralized documentation hub for GrowQR team
34 KiB
34 KiB
Phase 1: Dynamic Tool Registry Implementation
This document provides the concrete implementation plan for the first phase - replacing hardcoded routing maps with a dynamic tool registry.
Goals of Phase 1
- Replace
PAGE_TO_AGENTandACTION_TO_AGENThardcoded dictionaries with database lookups - Zero breaking changes to existing services
- Orchestrator behavior remains identical
- Foundation for runtime tool registration
Database Schema
-- tools table
CREATE TABLE tools (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
tool_id VARCHAR(255) UNIQUE NOT NULL, -- e.g., "resume_analyze"
name VARCHAR(255) NOT NULL,
description TEXT,
-- Service routing
service_name VARCHAR(255) NOT NULL, -- e.g., "resume-builder"
service_url VARCHAR(500), -- optional override
endpoint VARCHAR(255) DEFAULT '/a2a/tasks',
-- Triggers (how this tool is invoked)
page_triggers VARCHAR(255)[], -- e.g., ["resume-hub", "resume-editor"]
action_triggers VARCHAR(255)[], -- e.g., ["ai_analyze", "analyze_resume"]
skill_triggers VARCHAR(255)[], -- e.g., ["resume.analysis"]
-- JSON Schemas for validation
input_schema JSONB, -- JSON Schema for params
output_schema JSONB, -- JSON Schema for responses
-- Q-Score integration
qscore_signals VARCHAR(255)[], -- e.g., ["resume.ats_score"]
-- Versioning & metadata
version VARCHAR(50) DEFAULT '1.0.0',
enabled BOOLEAN DEFAULT true,
is_builtin BOOLEAN DEFAULT false, -- true for hardcoded tools
-- Ownership
created_by VARCHAR(255),
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
);
-- Index for fast lookups
CREATE INDEX idx_tools_page_triggers ON tools USING GIN(page_triggers);
CREATE INDEX idx_tools_action_triggers ON tools USING GIN(action_triggers);
CREATE INDEX idx_tools_service ON tools(service_name);
CREATE INDEX idx_tools_enabled ON tools(enabled) WHERE enabled = true;
-- Composite workflows table (for multi-step actions)
CREATE TABLE composite_workflows (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
workflow_id VARCHAR(255) UNIQUE NOT NULL,
name VARCHAR(255) NOT NULL,
description TEXT,
-- Steps as ordered JSON array
steps JSONB NOT NULL, -- [{"tool_id": "...", "params_mapping": {...}}]
action_trigger VARCHAR(255),
enabled BOOLEAN DEFAULT true,
created_at TIMESTAMP DEFAULT NOW()
);
New Files to Create
1. orchestrator/app/tools/registry.py
"""Dynamic tool registry - replaces hardcoded PAGE_TO_AGENT and ACTION_TO_AGENT."""
import logging
from dataclasses import dataclass
from typing import Optional
from datetime import datetime
import json
import asyncpg
from asyncpg import Pool
logger = logging.getLogger(__name__)
@dataclass
class Tool:
"""A registered tool that can be invoked by the orchestrator."""
id: str
tool_id: str
name: str
description: str
service_name: str
service_url: Optional[str]
endpoint: str
page_triggers: list[str]
action_triggers: list[str]
skill_triggers: list[str]
input_schema: Optional[dict]
output_schema: Optional[dict]
qscore_signals: list[str]
version: str
enabled: bool
is_builtin: bool
class ToolRegistry:
"""Database-backed tool registry with in-memory caching."""
def __init__(self, db_pool: Pool, cache_ttl_seconds: int = 30):
self._db = db_pool
self._cache_ttl = cache_ttl_seconds
self._cache: dict[str, Tool] = {} # tool_id -> Tool
self._page_index: dict[str, list[str]] = {} # page -> [tool_ids]
self._action_index: dict[str, list[str]] = {} # action -> [tool_ids]
self._last_refresh: Optional[datetime] = None
async def initialize(self):
"""Load all tools from database into cache."""
await self._refresh_cache()
logger.info(f"Tool registry initialized with {len(self._cache)} tools")
async def _refresh_cache(self):
"""Refresh the in-memory cache from database."""
async with self._db.acquire() as conn:
rows = await conn.fetch("""
SELECT * FROM tools WHERE enabled = true
""")
# Clear and rebuild
self._cache.clear()
self._page_index.clear()
self._action_index.clear()
for row in rows:
tool = Tool(
id=str(row['id']),
tool_id=row['tool_id'],
name=row['name'],
description=row['description'] or '',
service_name=row['service_name'],
service_url=row['service_url'],
endpoint=row['endpoint'],
page_triggers=row['page_triggers'] or [],
action_triggers=row['action_triggers'] or [],
skill_triggers=row['skill_triggers'] or [],
input_schema=row['input_schema'],
output_schema=row['output_schema'],
qscore_signals=row['qscore_signals'] or [],
version=row['version'],
enabled=row['enabled'],
is_builtin=row['is_builtin'],
)
self._cache[tool.tool_id] = tool
# Build indexes
for page in tool.page_triggers:
self._page_index.setdefault(page, []).append(tool.tool_id)
for action in tool.action_triggers:
self._action_index.setdefault(action, []).append(tool.tool_id)
self._last_refresh = datetime.utcnow()
async def get_tool_for_page(self, page: str) -> Optional[Tool]:
"""Find the primary tool for a given page."""
await self._maybe_refresh()
tool_ids = self._page_index.get(page, [])
if not tool_ids:
return None
# Return first enabled tool (or prioritize by version/is_builtin)
for tool_id in tool_ids:
tool = self._cache.get(tool_id)
if tool and tool.enabled:
return tool
return None
async def get_tool_for_action(self, action: str) -> Optional[Tool]:
"""Find the primary tool for a given action."""
await self._maybe_refresh()
# Exact match first
tool_ids = self._action_index.get(action, [])
if tool_ids:
for tool_id in tool_ids:
tool = self._cache.get(tool_id)
if tool and tool.enabled:
return tool
# Prefix match (e.g., "improve_headline" matches "improve_" prefix)
for trigger, tool_ids in self._action_index.items():
if trigger.endswith('_') and action.startswith(trigger):
for tool_id in tool_ids:
tool = self._cache.get(tool_id)
if tool and tool.enabled:
return tool
return None
async def get_service_for_page(self, page: str) -> Optional[str]:
"""Get service name for a page (backward compatible with old behavior)."""
tool = await self.get_tool_for_page(page)
return tool.service_name if tool else None
async def get_service_for_action(self, action: str) -> Optional[str]:
"""Get service name for an action (backward compatible)."""
tool = await self.get_tool_for_action(action)
return tool.service_name if tool else None
async def get_all_tools_for_page(self, page: str) -> list[Tool]:
"""Get all tools available for a page (for UI rendering)."""
await self._maybe_refresh()
tool_ids = self._page_index.get(page, [])
return [self._cache[tid] for tid in tool_ids if tid in self._cache]
async def list_tools(self, service_name: Optional[str] = None) -> list[Tool]:
"""List all tools, optionally filtered by service."""
await self._maybe_refresh()
tools = list(self._cache.values())
if service_name:
tools = [t for t in tools if t.service_name == service_name]
return tools
async def _maybe_refresh(self):
"""Refresh cache if TTL expired."""
if self._last_refresh is None:
await self._refresh_cache()
return
elapsed = (datetime.utcnow() - self._last_refresh).total_seconds()
if elapsed > self._cache_ttl:
await self._refresh_cache()
# Admin methods (for runtime registration)
async def register_tool(self, tool_data: dict) -> Tool:
"""Register a new tool at runtime."""
async with self._db.acquire() as conn:
row = await conn.fetchrow("""
INSERT INTO tools (
tool_id, name, description, service_name, service_url,
endpoint, page_triggers, action_triggers, skill_triggers,
input_schema, output_schema, qscore_signals,
version, enabled, is_builtin, created_by
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
ON CONFLICT (tool_id) DO UPDATE SET
name = EXCLUDED.name,
description = EXCLUDED.description,
service_name = EXCLUDED.service_name,
service_url = EXCLUDED.service_url,
endpoint = EXCLUDED.endpoint,
page_triggers = EXCLUDED.page_triggers,
action_triggers = EXCLUDED.action_triggers,
skill_triggers = EXCLUDED.skill_triggers,
input_schema = EXCLUDED.input_schema,
output_schema = EXCLUDED.output_schema,
qscore_signals = EXCLUDED.qscore_signals,
version = EXCLUDED.version,
updated_at = NOW()
RETURNING *
""",
tool_data['tool_id'],
tool_data['name'],
tool_data.get('description'),
tool_data['service_name'],
tool_data.get('service_url'),
tool_data.get('endpoint', '/a2a/tasks'),
tool_data.get('page_triggers', []),
tool_data.get('action_triggers', []),
tool_data.get('skill_triggers', []),
json.dumps(tool_data.get('input_schema')) if tool_data.get('input_schema') else None,
json.dumps(tool_data.get('output_schema')) if tool_data.get('output_schema') else None,
tool_data.get('qscore_signals', []),
tool_data.get('version', '1.0.0'),
tool_data.get('enabled', True),
tool_data.get('is_builtin', False),
tool_data.get('created_by', 'system')
)
await self._refresh_cache()
return Tool(
id=str(row['id']),
tool_id=row['tool_id'],
name=row['name'],
description=row['description'] or '',
service_name=row['service_name'],
service_url=row['service_url'],
endpoint=row['endpoint'],
page_triggers=row['page_triggers'] or [],
action_triggers=row['action_triggers'] or [],
skill_triggers=row['skill_triggers'] or [],
input_schema=row['input_schema'],
output_schema=row['output_schema'],
qscore_signals=row['qscore_signals'] or [],
version=row['version'],
enabled=row['enabled'],
is_builtin=row['is_builtin'],
)
async def unregister_tool(self, tool_id: str) -> bool:
"""Disable a tool (soft delete)."""
async with self._db.acquire() as conn:
result = await conn.execute("""
UPDATE tools SET enabled = false WHERE tool_id = $1
""", tool_id)
await self._refresh_cache()
return result == "UPDATE 1"
2. Migration Script: orchestrator/migrations/001_seed_builtin_tools.py
"""Seed the tools table with current hardcoded mappings."""
import asyncio
import asyncpg
from app.config import get_settings
# Copy of current hardcoded mappings from session.py
BUILTIN_TOOLS = [
# Dashboard
{
"tool_id": "home_dashboard",
"name": "Home Dashboard",
"description": "Main user dashboard with Q-Score and activity",
"service_name": "dashboard-service",
"page_triggers": ["home"],
"action_triggers": [],
"qscore_signals": ["dashboard.engagement"],
"is_builtin": True,
},
# Resume builder tools
{
"tool_id": "resume_page",
"name": "Resume Hub",
"description": "Resume management hub page",
"service_name": "resume-builder",
"page_triggers": ["resume-hub", "resume-editor", "cover-letter-editor"],
"action_triggers": [],
"is_builtin": True,
},
{
"tool_id": "resume_delete",
"name": "Delete Resume",
"service_name": "resume-builder",
"page_triggers": [],
"action_triggers": ["delete_resume"],
"is_builtin": True,
},
{
"tool_id": "resume_set_primary",
"name": "Set Primary Resume",
"service_name": "resume-builder",
"page_triggers": [],
"action_triggers": ["set_primary", "remove_primary"],
"is_builtin": True,
},
{
"tool_id": "resume_create",
"name": "Create Resume",
"service_name": "resume-builder",
"page_triggers": [],
"action_triggers": ["create_resume"],
"is_builtin": True,
},
{
"tool_id": "resume_parse",
"name": "Parse Resume",
"service_name": "resume-builder",
"page_triggers": [],
"action_triggers": ["parse_resume"],
"is_builtin": True,
},
{
"tool_id": "resume_analyze",
"name": "AI Resume Analysis",
"service_name": "resume-builder",
"page_triggers": [],
"action_triggers": ["ai_analyze"],
"qscore_signals": ["resume.ats_compatibility", "resume.completeness"],
"is_builtin": True,
},
{
"tool_id": "resume_copilot",
"name": "Resume AI Copilot",
"service_name": "resume-builder",
"page_triggers": [],
"action_triggers": ["ai_copilot", "ai_optimize_summary", "ai_optimize_experience",
"ai_suggest_skills", "ai_generate_summary"],
"is_builtin": True,
},
{
"tool_id": "resume_export",
"name": "Export Resume",
"service_name": "resume-builder",
"page_triggers": [],
"action_triggers": ["export_pdf", "export_analysis_pdf", "save_version", "update_resume_meta"],
"is_builtin": True,
},
# Cover letter tools
{
"tool_id": "cover_letter_generate",
"name": "Generate Cover Letter",
"service_name": "resume-builder",
"page_triggers": [],
"action_triggers": ["generate_cover_letter", "cover_letter_copilot"],
"is_builtin": True,
},
{
"tool_id": "cover_letter_manage",
"name": "Manage Cover Letter",
"service_name": "resume-builder",
"page_triggers": [],
"action_triggers": ["delete_cover_letter", "save_cover_letter_version",
"update_cover_letter_meta", "analyze_cover_letter", "export_cover_letter_pdf"],
"is_builtin": True,
},
# Social branding tools
{
"tool_id": "social_page",
"name": "Social Media Page",
"service_name": "social-branding",
"page_triggers": ["social-media"],
"action_triggers": [],
"is_builtin": True,
},
{
"tool_id": "social_submit_url",
"name": "Submit Social Profile URL",
"service_name": "social-branding",
"page_triggers": [],
"action_triggers": ["submit_url"],
"is_builtin": True,
},
{
"tool_id": "social_sync",
"name": "Sync Social Profile",
"service_name": "social-branding",
"page_triggers": [],
"action_triggers": ["sync_profile", "remove_profile"],
"is_builtin": True,
},
{
"tool_id": "social_analyze",
"name": "Analyze Social Profile",
"service_name": "social-branding",
"page_triggers": [],
"action_triggers": ["run_analysis"],
"qscore_signals": ["linkedin.headline_quality", "linkedin.profile_completeness"],
"is_builtin": True,
},
{
"tool_id": "social_improve",
"name": "Improve Social Profile",
"service_name": "social-branding",
"page_triggers": [],
"action_triggers": ["improve_"], # prefix match
"is_builtin": True,
},
{
"tool_id": "social_strategy",
"name": "Social Strategy",
"service_name": "social-branding",
"page_triggers": [],
"action_triggers": ["quick_wins", "generate_strategy", "generate_content_strategy"],
"is_builtin": True,
},
{
"tool_id": "social_content_delete",
"name": "Delete Social Content",
"service_name": "social-branding",
"page_triggers": [],
"action_triggers": ["delete_content"],
"is_builtin": True,
},
# Matchmaking tools
{
"tool_id": "matchmaking_page",
"name": "Job Matching Page",
"service_name": "matchmaking-service",
"page_triggers": ["job-matching"],
"action_triggers": [],
"is_builtin": True,
},
{
"tool_id": "matchmaking_sync",
"name": "Sync Matchmaking Preferences",
"service_name": "matchmaking-service",
"page_triggers": [],
"action_triggers": ["sync_preferences"],
"is_builtin": True,
},
{
"tool_id": "matchmaking_feed",
"name": "Get Job Feed",
"service_name": "matchmaking-service",
"page_triggers": [],
"action_triggers": ["get_feed"],
"qscore_signals": ["matching.feed_active"],
"is_builtin": True,
},
{
"tool_id": "matchmaking_feedback",
"name": "Record Job Feedback",
"service_name": "matchmaking-service",
"page_triggers": [],
"action_triggers": ["record_feedback", "get_opportunity_detail"],
"is_builtin": True,
},
# User service tools
{
"tool_id": "user_page",
"name": "User Profile Page",
"service_name": "user-service",
"page_triggers": ["profile", "settings"],
"action_triggers": ["view_profile"],
"is_builtin": True,
},
{
"tool_id": "user_update",
"name": "Update User",
"service_name": "user-service",
"page_triggers": [],
"action_triggers": ["update_preferences", "update_plan", "update_metadata"],
"is_builtin": True,
},
{
"tool_id": "user_qr",
"name": "QR Code Management",
"service_name": "user-service",
"page_triggers": [],
"action_triggers": ["get_qr_code", "regenerate_qr"],
"is_builtin": True,
},
# Interview service tools
{
"tool_id": "interview_page",
"name": "Interview Page",
"service_name": "interview-service",
"page_triggers": ["interview"],
"action_triggers": [],
"is_builtin": True,
},
{
"tool_id": "interview_configure",
"name": "Configure Interview",
"service_name": "interview-service",
"page_triggers": [],
"action_triggers": ["configure_interview"],
"is_builtin": True,
},
{
"tool_id": "interview_review",
"name": "Get Interview Review",
"service_name": "interview-service",
"page_triggers": [],
"action_triggers": ["get_review", "get_session_history"],
"qscore_signals": ["interview.overall_score"],
"is_builtin": True,
},
# Roleplay service tools
{
"tool_id": "roleplay_page",
"name": "Roleplay Page",
"service_name": "roleplay-service",
"page_triggers": ["roleplay"],
"action_triggers": [],
"is_builtin": True,
},
{
"tool_id": "roleplay_configure",
"name": "Configure Roleplay",
"service_name": "roleplay-service",
"page_triggers": [],
"action_triggers": ["configure_roleplay"],
"is_builtin": True,
},
{
"tool_id": "roleplay_review",
"name": "Get Roleplay Review",
"service_name": "roleplay-service",
"page_triggers": [],
"action_triggers": ["get_roleplay_review", "get_roleplay_history"],
"qscore_signals": ["roleplay.communication_score"],
"is_builtin": True,
},
# Q-Score tools
{
"tool_id": "qscore_get",
"name": "Get Q-Score",
"service_name": "qscore-service",
"page_triggers": [],
"action_triggers": ["get_qscore", "get_signals", "get_registry"],
"is_builtin": True,
},
{
"tool_id": "qscore_compute",
"name": "Compute Q-Score",
"service_name": "qscore-service",
"page_triggers": [],
"action_triggers": ["compute_qscore"],
"is_builtin": True,
},
# Pathways tools
{
"tool_id": "pathways_page",
"name": "Pathways Page",
"service_name": "pathways-service",
"page_triggers": ["pathways", "pathways-questionnaire", "pathways-generate", "pathways-weekly-plan"],
"action_triggers": [],
"is_builtin": True,
},
{
"tool_id": "pathways_profile",
"name": "Ingest Pathways Profile",
"service_name": "pathways-service",
"page_triggers": [],
"action_triggers": ["ingest_profile"],
"is_builtin": True,
},
{
"tool_id": "pathways_questionnaire",
"name": "Save Questionnaire",
"service_name": "pathways-service",
"page_triggers": [],
"action_triggers": ["save_questionnaire"],
"is_builtin": True,
},
{
"tool_id": "pathways_generate",
"name": "Generate Pathway",
"service_name": "pathways-service",
"page_triggers": [],
"action_triggers": ["generate_pathway", "activate_pathway", "get_pathway", "get_report"],
"qscore_signals": ["pathway.completion_pct"],
"is_builtin": True,
},
{
"tool_id": "pathways_tasks",
"name": "Pathway Task Management",
"service_name": "pathways-service",
"page_triggers": [],
"action_triggers": ["complete_task", "uncomplete_task", "get_weekly_plan"],
"is_builtin": True,
},
]
async def migrate():
settings = get_settings()
pool = await asyncpg.create_pool(settings.database_url)
async with pool.acquire() as conn:
# Create tables
await conn.execute("""
CREATE TABLE IF NOT EXISTS tools (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
tool_id VARCHAR(255) UNIQUE NOT NULL,
name VARCHAR(255) NOT NULL,
description TEXT,
service_name VARCHAR(255) NOT NULL,
service_url VARCHAR(500),
endpoint VARCHAR(255) DEFAULT '/a2a/tasks',
page_triggers VARCHAR(255)[],
action_triggers VARCHAR(255)[],
skill_triggers VARCHAR(255)[],
input_schema JSONB,
output_schema JSONB,
qscore_signals VARCHAR(255)[],
version VARCHAR(50) DEFAULT '1.0.0',
enabled BOOLEAN DEFAULT true,
is_builtin BOOLEAN DEFAULT false,
created_by VARCHAR(255),
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
)
""")
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_tools_page_triggers ON tools USING GIN(page_triggers)
""")
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_tools_action_triggers ON tools USING GIN(action_triggers)
""")
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_tools_service ON tools(service_name)
""")
# Insert builtin tools
for tool in BUILTIN_TOOLS:
await conn.execute("""
INSERT INTO tools (
tool_id, name, description, service_name,
page_triggers, action_triggers, skill_triggers, qscore_signals,
is_builtin, enabled
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
ON CONFLICT (tool_id) DO NOTHING
""",
tool['tool_id'],
tool['name'],
tool.get('description'),
tool['service_name'],
tool.get('page_triggers', []),
tool.get('action_triggers', []),
tool.get('skill_triggers', []),
tool.get('qscore_signals', []),
True, # is_builtin
True, # enabled
)
count = await conn.fetchval("SELECT COUNT(*) FROM tools WHERE is_builtin = true")
print(f"Seeded {count} builtin tools")
await pool.close()
if __name__ == "__main__":
asyncio.run(migrate())
3. Modified orchestrator/app/agent/session.py
Key changes to existing session.py (showing only the modified parts):
# REMOVE these hardcoded dictionaries:
# PAGE_TO_AGENT = {...}
# ACTION_TO_AGENT = {...}
# REPLACE with:
from app.tools.registry import ToolRegistry
class SuperAgentSession:
def __init__(
self,
websocket: WebSocket,
user_id: str,
registry: AgentRegistry,
tool_registry: ToolRegistry, # NEW parameter
a2a_client: A2AClient,
# ... rest of params
):
# ... existing init
self.tool_registry = tool_registry # NEW
# REMOVE these functions:
# def _resolve_agent_for_page(page, registry)
# def _resolve_agent_for_action(action, registry)
# def _resolve_agent_name_for_page(page)
# def _resolve_agent_name_for_action(action, registry)
# REPLACE with async versions using tool registry:
async def _resolve_agent_for_page(self, page: str):
"""Find the healthy agent responsible for a page using tool registry."""
tool = await self.tool_registry.get_tool_for_page(page)
if not tool:
return None
agent = self.registry.get_agent_by_name(tool.service_name)
if agent and agent.is_healthy:
return agent
return None
async def _resolve_agent_for_action(self, action: str):
"""Find the healthy agent responsible for an action using tool registry."""
tool = await self.tool_registry.get_tool_for_action(action)
if not tool:
return None
agent = self.registry.get_agent_by_name(tool.service_name)
if agent and agent.is_healthy:
return agent
return None
async def _resolve_agent_name_for_page(self, page: str) -> str | None:
"""Return the agent name for a page using tool registry."""
return await self.tool_registry.get_service_for_page(page)
async def _resolve_agent_name_for_action(self, action: str) -> str | None:
"""Return the agent name for an action using tool registry."""
return await self.tool_registry.get_service_for_action(action)
# UPDATE on_session_start and on_user_action to be async and use new methods
async def on_session_start(self, page: str, params: dict):
"""Route session_start to the appropriate domain agent."""
if self._use_redis:
agent_name = await self._resolve_agent_name_for_page(page) # NOW ASYNC
# ... rest of method
else:
agent = await self._resolve_agent_for_page(page) # NOW ASYNC
# ... rest of method
async def on_user_action(self, action: str, params: dict):
"""Route action to domain agent or composite workflow."""
# ... composite check stays same
if self._use_redis:
agent_name = await self._resolve_agent_name_for_action(action) # NOW ASYNC
# ... rest of method
else:
agent = await self._resolve_agent_for_action(action) # NOW ASYNC
# ... rest of method
4. New API Endpoints: orchestrator/app/api/v1/tools.py
"""Admin API for tool registry management."""
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from typing import Optional, List
from app.tools.registry import ToolRegistry
from app.auth import require_admin # You'll need to implement this
router = APIRouter()
class ToolRegistration(BaseModel):
tool_id: str
name: str
description: Optional[str] = None
service_name: str
service_url: Optional[str] = None
endpoint: str = "/a2a/tasks"
page_triggers: List[str] = []
action_triggers: List[str] = []
skill_triggers: List[str] = []
input_schema: Optional[dict] = None
output_schema: Optional[dict] = None
qscore_signals: List[str] = []
version: str = "1.0.0"
class ToolResponse(BaseModel):
id: str
tool_id: str
name: str
service_name: str
enabled: bool
version: str
@router.post("/register", response_model=ToolResponse)
async def register_tool(
registration: ToolRegistration,
registry: ToolRegistry = Depends(get_tool_registry),
admin = Depends(require_admin),
):
"""Register a new tool dynamically.
This allows services to self-register their capabilities at runtime.
"""
tool = await registry.register_tool(registration.dict())
return ToolResponse(
id=tool.id,
tool_id=tool.tool_id,
name=tool.name,
service_name=tool.service_name,
enabled=tool.enabled,
version=tool.version,
)
@router.post("/{tool_id}/unregister")
async def unregister_tool(
tool_id: str,
registry: ToolRegistry = Depends(get_tool_registry),
admin = Depends(require_admin),
):
"""Disable a tool (soft delete)."""
success = await registry.unregister_tool(tool_id)
if not success:
raise HTTPException(status_code=404, detail="Tool not found")
return {"success": True}
@router.get("/available")
async def get_available_tools(
page: Optional[str] = None,
service: Optional[str] = None,
registry: ToolRegistry = Depends(get_tool_registry),
):
"""Get available tools for a page or service.
Used by frontend to discover what tools are available on a given page.
"""
if page:
tools = await registry.get_all_tools_for_page(page)
else:
tools = await registry.list_tools(service_name=service)
return {
"tools": [
{
"tool_id": t.tool_id,
"name": t.name,
"description": t.description,
"service_name": t.service_name,
"action_triggers": t.action_triggers,
"qscore_signals": t.qscore_signals,
"version": t.version,
}
for t in tools if t.enabled
]
}
@router.get("/{tool_id}")
async def get_tool_details(
tool_id: str,
registry: ToolRegistry = Depends(get_tool_registry),
):
"""Get detailed information about a specific tool."""
# Implementation to fetch single tool details
pass
def get_tool_registry() -> ToolRegistry:
"""Dependency to get the tool registry instance."""
# This should return the singleton instance from app state
from app.main import app
return app.state.tool_registry
Configuration Changes
orchestrator/app/config.py
Add database URL for tool registry:
class Settings(BaseSettings):
# ... existing settings
# Tool registry database (can be same as other services or separate)
tool_registry_database_url: str = os.getenv(
"TOOL_REGISTRY_DATABASE_URL",
os.getenv("DATABASE_URL", "postgresql://user:pass@localhost/growqr_tools")
)
# Cache TTL for tool registry
tool_registry_cache_ttl: int = int(os.getenv("TOOL_REGISTRY_CACHE_TTL", "30"))
orchestrator/app/main.py
Initialize tool registry on startup:
from app.tools.registry import ToolRegistry
@app.on_event("startup")
async def startup():
# ... existing startup code
# Initialize tool registry
db_pool = await asyncpg.create_pool(settings.tool_registry_database_url)
tool_registry = ToolRegistry(db_pool, cache_ttl_seconds=settings.tool_registry_cache_ttl)
await tool_registry.initialize()
app.state.tool_registry = tool_registry
app.state.db_pool = db_pool
@app.on_event("shutdown")
async def shutdown():
# ... existing shutdown code
if hasattr(app.state, 'db_pool'):
await app.state.db_pool.close()
Migration Steps (Zero Downtime)
-
Pre-deploy: Database Setup
# Run migration script to create tables and seed builtin tools cd orchestrator python migrations/001_seed_builtin_tools.py -
Deploy: New Orchestrator Version
- Deploy orchestrator with tool registry changes
- New code reads from database but falls back to hardcoded if needed
- Old services continue working unchanged
-
Verify: Check Registry Loaded
curl http://orchestrator:8000/api/v1/tools/available?page=home # Should return dashboard-service tool -
Post-deploy: Remove Hardcoded Fallbacks
- After verification, deploy version 2 that removes hardcoded dictionaries entirely
Benefits Achieved in Phase 1
- Routing is now data-driven: Change page/action routing by updating database, not code
- No orchestrator redeploy needed for new tool registrations
- Original team can add tools via API without your involvement
- Your service changes don't break routing as long as service_name stays consistent
- Foundation for Phase 2: Self-registration becomes trivial
- Audit trail: Database records who created what tool and when
Next Steps After Phase 1
- Phase 2: Add service self-registration (services call
/api/v1/tools/registeron startup) - Phase 3: Add frontend tool discovery API for dynamic UI rendering
- Phase 4: Create lightweight SDK for contributors