# 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 1. Replace `PAGE_TO_AGENT` and `ACTION_TO_AGENT` hardcoded dictionaries with database lookups 2. Zero breaking changes to existing services 3. Orchestrator behavior remains identical 4. Foundation for runtime tool registration ## Database Schema ```sql -- 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` ```python """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` ```python """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): ```python # 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` ```python """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: ```python 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: ```python 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) 1. **Pre-deploy: Database Setup** ```bash # Run migration script to create tables and seed builtin tools cd orchestrator python migrations/001_seed_builtin_tools.py ``` 2. **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 3. **Verify: Check Registry Loaded** ```bash curl http://orchestrator:8000/api/v1/tools/available?page=home # Should return dashboard-service tool ``` 4. **Post-deploy: Remove Hardcoded Fallbacks** - After verification, deploy version 2 that removes hardcoded dictionaries entirely ## Benefits Achieved in Phase 1 1. **Routing is now data-driven**: Change page/action routing by updating database, not code 2. **No orchestrator redeploy needed** for new tool registrations 3. **Original team can add tools** via API without your involvement 4. **Your service changes** don't break routing as long as service_name stays consistent 5. **Foundation for Phase 2**: Self-registration becomes trivial 6. **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/register` on startup) - Phase 3: Add frontend tool discovery API for dynamic UI rendering - Phase 4: Create lightweight SDK for contributors