""" Tool sync module for tfcode. Syncs tools from ToothFairyAI workspace using the official SDK. SDK Structure: - agent_functions: API Functions (with request_type) - connections: Provider connections (openai, anthropic, etc.) - agents: TF workspace agents - prompts: Prompt templates (with available_to_agents mapping) """ from typing import Any, Optional, List from pydantic import BaseModel from toothfairyai.types import AgentFunction from tf_sync.config import TFConfig, ToolType, FunctionRequestType class SyncedTool(BaseModel): """A tool synced from ToothFairyAI workspace.""" id: str name: str description: Optional[str] = None tool_type: ToolType is_mcp_server: bool = False is_agent_skill: bool = False is_database_script: bool = False request_type: Optional[FunctionRequestType] = None url: Optional[str] = None tools: list[str] = [] authorisation_type: Optional[str] = None auth_via: str = "tf_proxy" # Coder agent specific fields for prompting/model configuration interpolation_string: Optional[str] = None goals: Optional[str] = None temperature: Optional[float] = None max_tokens: Optional[int] = None llm_base_model: Optional[str] = None llm_provider: Optional[str] = None class SyncedPrompt(BaseModel): """A prompt template synced from ToothFairyAI workspace.""" id: str label: str interpolation_string: str prompt_type: Optional[str] = None available_to_agents: Optional[List[str]] = None description: Optional[str] = None class ToolSyncResult(BaseModel): """Result of tool sync operation.""" success: bool tools: list[SyncedTool] = [] prompts: list[SyncedPrompt] = [] by_type: dict[str, int] = {} error: Optional[str] = None def classify_tool(func: AgentFunction) -> ToolType: """ Classify a tool based on its properties. Types: - AGENT_SKILL: is_agent_skill=True - API_FUNCTION: has request_type Args: func: AgentFunction from TF SDK Returns: ToolType enum value """ # Agent skills have is_agent_skill=True if getattr(func, 'is_agent_skill', None) is True: return ToolType.AGENT_SKILL # All agent_functions with request_type are API Functions if func.request_type: return ToolType.API_FUNCTION return ToolType.API_FUNCTION def parse_function(func: AgentFunction) -> SyncedTool: """ Parse AgentFunction from SDK into SyncedTool. Args: func: AgentFunction from TF SDK Returns: SyncedTool instance """ tool_type = classify_tool(func) request_type_enum = None if func.request_type: try: request_type_enum = FunctionRequestType(func.request_type) except ValueError: pass # API Functions may have user-provided auth (authorisation_type) # or may use TF proxy auth_via = "user_provided" if func.authorisation_type == "api_key" else "tf_proxy" # Agent skills use skill script if tool_type == ToolType.AGENT_SKILL: auth_via = "tf_skill" return SyncedTool( id=func.id, name=func.name, description=func.description, tool_type=tool_type, request_type=request_type_enum, url=func.url, authorisation_type=func.authorisation_type, auth_via=auth_via, is_agent_skill=tool_type == ToolType.AGENT_SKILL, ) def parse_agent(agent) -> SyncedTool: """ Parse Agent from SDK into SyncedTool. Coder agents (mode='coder') are CODER_AGENT type, not skills. Args: agent: Agent from TF SDK Returns: SyncedTool instance with full agent configuration """ return SyncedTool( id=agent.id, name=agent.label or f"agent_{agent.id[:8]}", description=agent.description, tool_type=ToolType.CODER_AGENT, is_agent_skill=False, auth_via="tf_agent", # Agent prompting configuration interpolation_string=getattr(agent, 'interpolation_string', None), goals=getattr(agent, 'goals', None), # Agent model configuration temperature=getattr(agent, 'temperature', None), max_tokens=getattr(agent, 'max_tokens', None), llm_base_model=getattr(agent, 'llm_base_model', None), llm_provider=getattr(agent, 'llm_provider', None), ) def parse_prompt(prompt) -> SyncedPrompt: """ Parse Prompt from SDK into SyncedPrompt. Args: prompt: Prompt from TF SDK Returns: SyncedPrompt instance """ return SyncedPrompt( id=prompt.id, label=prompt.label, interpolation_string=prompt.interpolation_string, prompt_type=getattr(prompt, 'prompt_type', None), available_to_agents=getattr(prompt, 'available_to_agents', None), description=getattr(prompt, 'description', None), ) def sync_tools(config: TFConfig) -> ToolSyncResult: """ Sync all tools from ToothFairyAI workspace using SDK. Includes: - Agent Functions (API Functions with request_type) - Agent Skills (functions with is_agent_skill=True) - Coder Agents (agents with mode='coder') - Prompts (prompt templates with available_to_agents mapping) Args: config: TFConfig instance Returns: ToolSyncResult with synced tools and prompts """ try: client = config.get_client() # Sync agent functions (API auto-paginates up to 5000) func_result = client.agent_functions.list() tools = [parse_function(f) for f in func_result.items] # Sync coder agents (API auto-paginates up to 5000) try: agents_result = client.agents.list() for agent in agents_result.items: if getattr(agent, 'mode', None) == 'coder': tools.append(parse_agent(agent)) except Exception: pass # Sync prompts (API auto-paginates up to 5000) prompts = [] try: prompts_result = client.prompts.list() prompts = [parse_prompt(p) for p in prompts_result.items] except Exception: pass by_type = {} for tool in tools: type_name = tool.tool_type.value by_type[type_name] = by_type.get(type_name, 0) + 1 if prompts: by_type['prompt'] = len(prompts) return ToolSyncResult( success=True, tools=tools, prompts=prompts, by_type=by_type, ) except Exception as e: return ToolSyncResult( success=False, error=f"Sync failed: {str(e)}", ) def sync_tools_by_type( config: TFConfig, tool_types: Optional[list[ToolType]] = None, ) -> ToolSyncResult: """ Sync tools of specific types from ToothFairyAI workspace. Args: config: TFConfig instance tool_types: List of ToolType to sync (None = all) Returns: ToolSyncResult with filtered tools """ result = sync_tools(config) if not result.success or not tool_types: return result filtered = [t for t in result.tools if t.tool_type in tool_types] by_type = {} for tool in filtered: type_name = tool.tool_type.value by_type[type_name] = by_type.get(type_name, 0) + 1 return ToolSyncResult( success=True, tools=filtered, by_type=by_type, ) def sync_api_functions_only(config: TFConfig) -> ToolSyncResult: """Sync only API Functions (has requestType).""" return sync_tools_by_type(config, [ToolType.API_FUNCTION])