""" Copyright 2024-2026 ReplyPier Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. """ import os from fastapi import APIRouter, Depends, HTTPException, status, UploadFile, File, Request from typing import List from app.core.logger import get_logger from app.database import get_db from app.models.user import User, UserGroup from app.core.auth import ( INBOX_PERMISSIONS, require_any_permission, require_any_unified_permission, require_permissions, require_unified_permissions, ) from app.repositories.agent import AgentRepository from app.repositories.knowledge import KnowledgeRepository from app.models.schemas.agent import AgentUpdate, AgentCreate, AgentRosterItem, AgentWithCustomizationResponse from sqlalchemy.orm import Session from app.models.agent import Agent, AgentCustomization from app.models.organization import Organization from app.agents.guardrail_policy import DEFAULT_GUARDRAIL_PROMPT from app.models.schemas.agent_customization import CustomizationCreate, CustomizationResponse import aiofiles from uuid import uuid4 from PIL import Image from uuid import UUID from app.core.s3 import upload_file_to_s3 from app.core.config import settings from app.services import telegram_webhooks from app.services.file_storage import local_upload_path from pydantic import BaseModel from agno.agent import Agent as AgnoAgent from app.utils.agno_utils import create_model from app.utils.rate_limit import limit_instruction_generation router = APIRouter() logger = get_logger(__name__) UPLOAD_DIR = "uploads/agents" ALLOWED_MIME_TYPES = {'image/jpeg', 'image/png', 'image/webp'} # The stored name's extension comes from the validated content type, never from # the caller's filename — nothing user-supplied reaches the path that way. MIME_EXTENSIONS = {'image/jpeg': '.jpg', 'image/png': '.png', 'image/webp': '.webp'} MAX_FILE_SIZE = 5 * 1024 * 1024 # 5MB # Model for instruction generation request class InstructionPrompt(BaseModel): prompt: str existing_instructions: List[str] = [] # Optional list of existing instructions # Model for setting active workflow class SetActiveWorkflowRequest(BaseModel): workflow_id: UUID use_workflow: bool = True # Helper functions for agent access control def verify_agent_access(agent: Agent, auth_info: dict) -> None: """Verify that the authenticated user/session has access to the agent""" # Compare UUIDs directly (auth_info["organization_id"] is now a UUID, not a string) if agent.organization_id != auth_info["organization_id"]: # Return 404 instead of 403 for security - don't reveal resource existence raise HTTPException( status_code=404, detail="Agent not found" ) async def save_file(file: UploadFile, organization_id: UUID) -> str: """Save uploaded file and return the file path""" # Validate file size file.file.seek(0, 2) # Seek to end size = file.file.tell() file.file.seek(0) # Reset file position if size > MAX_FILE_SIZE: raise HTTPException( status_code=400, detail=f"File size exceeds maximum limit of { MAX_FILE_SIZE // 1024 // 1024}MB" ) # Validate image type if file.content_type not in ALLOWED_MIME_TYPES: raise HTTPException( status_code=400, detail="Only JPEG, PNG and WebP images are allowed" ) try: # Verify it's a valid image img = Image.open(file.file) img.verify() file.file.seek(0) except Exception: raise HTTPException( status_code=400, detail="Invalid image file" ) file_name = f"{uuid4()}{MIME_EXTENSIONS[file.content_type]}" if settings.S3_FILE_STORAGE: folder = f"agents/{organization_id}" file.file.seek(0) content = await file.read() return await upload_file_to_s3(content, folder, file_name, content_type=file.content_type) else: # Local storage upload_dir = f"uploads/agents/{organization_id}" if not os.path.exists(upload_dir): os.makedirs(upload_dir) file_path = os.path.join(upload_dir, file_name) async with aiofiles.open(file_path, 'wb') as f: content = await file.read() await f.write(content) # The uploads/ directory is static-mounted at {API_V1_STR}/uploads, # so the stored URL must carry that prefix (same shape as store_upload). return f"{settings.API_V1_STR}/{file_path}" @router.post("", response_model=AgentWithCustomizationResponse, status_code=status.HTTP_201_CREATED) async def create_agent( agent_data: AgentCreate, current_user: User = Depends(require_permissions("manage_agents")), db: Session = Depends(get_db) ): """Create a new agent""" try: agent_repo = AgentRepository(db) # Check enterprise subscription limits if enterprise module is available try: from app.enterprise.services.feature_access import require_accessible_subscription HAS_ENTERPRISE = True except ImportError: HAS_ENTERPRISE = False if HAS_ENTERPRISE: # Accessible = active/trial/past-due-in-period OR cancelled-but- # still-in-paid-period; raises 403 when the org has no plan. subscription = require_accessible_subscription(db, current_user.organization_id) # Get current agents count current_agents_count = agent_repo.count_by_organization(current_user.organization_id) # Check if adding this agent would exceed the limit if subscription.plan.max_agents is not None and (current_agents_count + 1) > subscription.plan.max_agents: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"Cannot create agent: Maximum number of agents ({subscription.plan.max_agents}) exceeded for current plan , please upgrade your plan" ) # Check if agent with the same name already exists in the organization existing_agent = agent_repo.get_by_name(agent_data.name, current_user.organization_id) if existing_agent: raise HTTPException( status_code=400, detail=f"Agent with name '{agent_data.name}' already exists" ) # Create agent with organization ID from current user agent_dict = agent_data.model_dump() # Override organization_id with the current user's organization agent_dict["organization_id"] = current_user.organization_id # Set display_name to name if not provided if "display_name" not in agent_dict or not agent_dict.get("display_name"): agent_dict["display_name"] = agent_dict["name"] # Check rating feature availability and override ask_for_rating if necessary if HAS_ENTERPRISE: from app.enterprise.repositories.plan import PlanRepository plan_repo = PlanRepository(db) if not plan_repo.check_feature_availability(str(subscription.plan_id), 'rating'): agent_dict["ask_for_rating"] = False # Create the agent agent = agent_repo.create_agent(**agent_dict) # Prepare response with empty knowledge list response = AgentWithCustomizationResponse( id=agent.id, name=agent.name, display_name=agent.display_name, description=agent.description, agent_type=agent.agent_type, instructions=agent.instructions, is_active=agent.is_active, organization_id=agent.organization_id, customization=None, transfer_to_human=agent.transfer_to_human or False, ai_replies_enabled=agent.ai_replies_enabled, channel_quick_replies=agent.channel_quick_replies, ask_for_rating=agent.ask_for_rating or False, handoff_collect_email=agent.handoff_collect_email, handoff_collect_name=agent.handoff_collect_name, enable_rate_limiting=agent.enable_rate_limiting or False, overall_limit_per_ip=agent.overall_limit_per_ip or 100, requests_per_sec=agent.requests_per_sec or 1.0, use_workflow=agent.use_workflow or False, active_workflow_id=agent.active_workflow_id, allow_attachments=agent.allow_attachments or False, allowed_attachment_types=agent.allowed_attachment_types, require_token_auth=agent.require_token_auth or False, ticketing_enabled=agent.ticketing_enabled if agent.ticketing_enabled is not None else True, topic_scope=agent.topic_scope, guardrail_prompt=agent.guardrail_prompt, guardrail_enabled=agent.guardrail_enabled if agent.guardrail_enabled is not None else True, knowledge=[], groups=[] ) return response except HTTPException as e: # Re-raise HTTP exceptions raise e except Exception as e: logger.error(f"Error creating agent: {str(e)}") db.rollback() raise HTTPException(status_code=500, detail=str(e)) @router.put("/{agent_id}/shopify", response_model=AgentWithCustomizationResponse) @router.put("/{agent_id}", response_model=AgentWithCustomizationResponse) async def update_agent( agent_id: UUID, update_data: AgentUpdate, auth_info: dict = Depends(require_unified_permissions("manage_agents")), db: Session = Depends(get_db) ): """Update agent details - supports both JWT and Shopify session token auth""" try: agent_repo = AgentRepository(db) knowledge_repo = KnowledgeRepository(db) # Get existing agent agent = agent_repo.get_by_id(agent_id) if not agent: raise HTTPException(status_code=404, detail="Agent not found") # Verify organization access verify_agent_access(agent, auth_info) logger.info(f"Updating agent {agent_id} with data: {update_data}") # Check enterprise subscription limits if enterprise module is available try: from app.enterprise.repositories.subscription import SubscriptionRepository HAS_ENTERPRISE = True except ImportError: HAS_ENTERPRISE = False # Update agent with provided fields, excluding photo_url update_dict = update_data.model_dump( exclude={'photo_url'}, exclude_unset=True) # Check rating feature availability and override ask_for_rating if necessary. # Applies to both JWT and Personal Access Token (PAT) sessions; Shopify auth # enforces plan features separately. if HAS_ENTERPRISE and auth_info['auth_type'] in ('jwt', 'pat'): from app.enterprise.repositories.plan import PlanRepository subscription_repo = SubscriptionRepository(db) # Accessible sub (incl. cancelled-but-still-in-paid-period) so a # cancelled Pro keeps its rating feature until the period ends. subscription = subscription_repo.get_active_subscription(auth_info["organization_id"]) if subscription: plan_repo = PlanRepository(db) if not plan_repo.check_feature_availability(str(subscription.plan_id), 'rating'): # If rating feature is not available, force ask_for_rating to False update_dict["ask_for_rating"] = False buttons_turned_on = update_dict.get("channel_quick_replies") is True and not agent.channel_quick_replies agent = agent_repo.update_agent(agent_id, **update_dict) if buttons_turned_on: # Telegram bots connected before buttons existed don't receive taps yet await telegram_webhooks.reregister_for_agent(db, agent.id) # Get knowledge sources for response knowledge_items = knowledge_repo.get_by_agent(agent.id) # photo_url is signed by CustomizationResponse on serialization — do not # assign the signed value onto the ORM object, it gets flushed to the DB. customization = agent.customization # Prepare response response = AgentWithCustomizationResponse( id=agent.id, name=agent.name, display_name=agent.display_name, description=agent.description, agent_type=agent.agent_type, instructions=agent.instructions, is_active=agent.is_active, use_workflow=agent.use_workflow, active_workflow_id=agent.active_workflow_id, organization_id=agent.organization_id, customization=customization, transfer_to_human=agent.transfer_to_human, ai_replies_enabled=agent.ai_replies_enabled, channel_quick_replies=agent.channel_quick_replies, ask_for_rating=agent.ask_for_rating, handoff_collect_email=agent.handoff_collect_email, handoff_collect_name=agent.handoff_collect_name, enable_rate_limiting=agent.enable_rate_limiting, overall_limit_per_ip=agent.overall_limit_per_ip, requests_per_sec=agent.requests_per_sec, allow_attachments=agent.allow_attachments, allowed_attachment_types=agent.allowed_attachment_types, require_token_auth=agent.require_token_auth, ticketing_enabled=agent.ticketing_enabled if agent.ticketing_enabled is not None else True, topic_scope=agent.topic_scope, guardrail_prompt=agent.guardrail_prompt, guardrail_enabled=agent.guardrail_enabled if agent.guardrail_enabled is not None else True, knowledge=[{ "id": k.id, "name": k.source, "type": k.source_type } for k in knowledge_items], groups=agent.groups ) return response except HTTPException as e: # Re-raise HTTP exceptions raise e except Exception as e: logger.error(f"Agent update error: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) @router.get("/roster", response_model=List[AgentRosterItem]) async def get_agent_roster( current_user: User = Depends(require_any_permission(*INBOX_PERMISSIONS)), db: Session = Depends(get_db) ): """Names of the org's AI agents, for the inbox filter. Declared above GET /{agent_id}, which takes a UUID — below it, "roster" fails UUID parsing and 422s. Deliberately not GET /agent/list: AgentResponse carries `instructions` and `guardrail_prompt`, and the guardrail prompt tells a reader exactly how to phrase a bypass. Filtering conversations needs a name. """ agents = AgentRepository(db).get_all_agents(current_user.organization_id) return agents @router.get("/guardrail-default") async def get_guardrail_default( auth_info: dict = Depends( require_any_unified_permission("view_agents", "manage_agents")), db: Session = Depends(get_db) ): """The shipped scope rule, with {org} already filled in. The dashboard shows this as the starting text so operators can read and edit the actual rule instead of a prose summary of it — which drifted from the real wording the first time the default changed. Served from the backend rather than duplicated in the frontend so there is one source of truth. Declared before /{agent_id} so the literal path wins the route match. """ org = db.query(Organization).filter( Organization.id == auth_info["organization_id"] ).first() label = (org.name or "").strip() if org else "" return {"prompt": DEFAULT_GUARDRAIL_PROMPT.replace("{org}", label or "this business")} @router.get("/list/shopify", response_model=List[AgentWithCustomizationResponse]) @router.get("/list", response_model=List[AgentWithCustomizationResponse]) async def get_organization_agents( auth_info: dict = Depends( require_any_unified_permission("view_agents", "manage_agents")), db: Session = Depends(get_db) ): """Get organization agents - supports both JWT and Shopify session token auth""" try: agent_repo = AgentRepository(db) knowledge_repo = KnowledgeRepository(db) agents = agent_repo.get_all_agents(auth_info["organization_id"]) response = [] for agent in agents: knowledge_items = knowledge_repo.get_by_agent(agent.id) # photo_url is signed by CustomizationResponse on serialization. customization = agent.customization agent_data = AgentWithCustomizationResponse( id=agent.id, name=agent.name, display_name=agent.display_name, description=agent.description, agent_type=agent.agent_type, instructions=agent.instructions, is_active=agent.is_active, organization_id=agent.organization_id, transfer_to_human=agent.transfer_to_human or False, ai_replies_enabled=agent.ai_replies_enabled, channel_quick_replies=agent.channel_quick_replies, ask_for_rating=agent.ask_for_rating or False, handoff_collect_email=agent.handoff_collect_email, handoff_collect_name=agent.handoff_collect_name, enable_rate_limiting=agent.enable_rate_limiting or False, overall_limit_per_ip=agent.overall_limit_per_ip or 100, requests_per_sec=agent.requests_per_sec or 1.0, use_workflow=agent.use_workflow or False, active_workflow_id=agent.active_workflow_id, created_at=agent.created_at, updated_at=agent.updated_at, allow_attachments=agent.allow_attachments or False, allowed_attachment_types=agent.allowed_attachment_types, require_token_auth=agent.require_token_auth or False, ticketing_enabled=agent.ticketing_enabled if agent.ticketing_enabled is not None else True, topic_scope=agent.topic_scope, guardrail_prompt=agent.guardrail_prompt, guardrail_enabled=agent.guardrail_enabled if agent.guardrail_enabled is not None else True, knowledge=[{ "id": k.id, "name": k.source, "type": k.source_type } for k in knowledge_items], customization=customization, groups=agent.groups ) response.append(agent_data) return response except Exception as e: logger.error(f"Error getting agents: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) @router.post("/{agent_id}/customization/shopify", response_model=CustomizationResponse) @router.post("/{agent_id}/customization", response_model=CustomizationResponse) async def create_agent_customization( agent_id: UUID, customization_data: CustomizationCreate, auth_info: dict = Depends(require_unified_permissions("manage_agents")), db: Session = Depends(get_db) ): """Create or update agent customization - supports both JWT and Shopify session token auth""" try: # Get agent agent = db.query(Agent).filter(Agent.id == agent_id).first() if not agent: raise HTTPException(status_code=404, detail="Agent not found") # Verify organization access verify_agent_access(agent, auth_info) # Get existing customization or create new one db_customization = db.query(AgentCustomization).filter( AgentCustomization.agent_id == agent_id ).first() if db_customization: # Update existing customization for field, value in customization_data.model_dump(exclude_unset=True).items(): if field != 'id': # Don't update the ID setattr(db_customization, field, value) else: # Create new customization db_customization = AgentCustomization( agent_id=agent_id, **customization_data.model_dump(exclude={'id'}) ) db.add(db_customization) db.commit() db.refresh(db_customization) # photo_url is signed by CustomizationResponse on serialization. Assigning # it here would leave the refreshed instance dirty and persist the # signature on the next flush. return db_customization except HTTPException as e: # Re-raise HTTP exceptions raise e except Exception as e: logger.error(f"Error creating agent customization: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) @router.post("/{agent_id}/customization/photo/shopify", response_model=CustomizationResponse) @router.post("/{agent_id}/customization/photo", response_model=CustomizationResponse) async def upload_agent_photo( agent_id: str, photo: UploadFile = File(...), auth_info: dict = Depends(require_unified_permissions("manage_agents")), db: Session = Depends(get_db) ): """Upload agent profile photo - supports both JWT and Shopify session token auth""" try: agent = db.query(Agent).filter(Agent.id == agent_id).first() if not agent: raise HTTPException(status_code=404, detail="Agent not found") # Verify organization access verify_agent_access(agent, auth_info) # Get existing customization to check for old photo db_customization = db.query(AgentCustomization).filter( AgentCustomization.agent_id == agent_id ).first() # Delete old photo if it exists if db_customization and db_customization.photo_url: if settings.S3_FILE_STORAGE: from app.core.s3 import delete_file_from_s3 await delete_file_from_s3(db_customization.photo_url) else: old_photo_path = local_upload_path(db_customization.photo_url) if os.path.exists(old_photo_path): os.remove(old_photo_path) # Save new photo file photo_url = await save_file(photo, auth_info["organization_id"]) # Update or create customization if db_customization: db_customization.photo_url = photo_url else: db_customization = AgentCustomization( agent_id=agent_id, photo_url=photo_url ) db.add(db_customization) db.commit() db.refresh(db_customization) # photo_url is signed by CustomizationResponse on serialization. Assigning # it here would leave the refreshed instance dirty and persist the # signature on the next flush. return db_customization except HTTPException: raise except Exception as e: logger.error(f"Error uploading agent photo: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to upload photo" ) @router.put("/{agent_id}/groups", response_model=AgentWithCustomizationResponse) async def update_agent_groups( agent_id: UUID, group_ids: List[UUID], current_user: User = Depends(require_permissions("manage_agents")), db: Session = Depends(get_db) ): """Update agent's assigned groups""" try: # Get agent agent = db.query(Agent).filter(Agent.id == agent_id).first() if not agent or agent.organization_id != current_user.organization_id: raise HTTPException(status_code=404, detail="Agent not found") # Verify all groups exist and belong to same organization groups = db.query(UserGroup).filter( UserGroup.id.in_(group_ids), UserGroup.organization_id == current_user.organization_id ).all() if len(groups) != len(group_ids): raise HTTPException(status_code=400, detail="Invalid group IDs provided") # Update agent's groups agent.groups = groups db.commit() db.refresh(agent) # Prepare response knowledge_repo = KnowledgeRepository(db) knowledge_items = knowledge_repo.get_by_agent(agent.id) customization = agent.customization return AgentWithCustomizationResponse( id=agent.id, name=agent.name, display_name=agent.display_name, description=agent.description, agent_type=agent.agent_type, instructions=agent.instructions, is_active=agent.is_active, use_workflow=agent.use_workflow, active_workflow_id=agent.active_workflow_id, organization_id=agent.organization_id, customization=customization, transfer_to_human=agent.transfer_to_human, ai_replies_enabled=agent.ai_replies_enabled, channel_quick_replies=agent.channel_quick_replies, ask_for_rating=agent.ask_for_rating, handoff_collect_email=agent.handoff_collect_email, handoff_collect_name=agent.handoff_collect_name, enable_rate_limiting=agent.enable_rate_limiting, overall_limit_per_ip=agent.overall_limit_per_ip, requests_per_sec=agent.requests_per_sec, allow_attachments=agent.allow_attachments, allowed_attachment_types=agent.allowed_attachment_types, require_token_auth=agent.require_token_auth, ticketing_enabled=agent.ticketing_enabled if agent.ticketing_enabled is not None else True, topic_scope=agent.topic_scope, guardrail_prompt=agent.guardrail_prompt, guardrail_enabled=agent.guardrail_enabled if agent.guardrail_enabled is not None else True, groups=agent.groups, knowledge=[{ "id": k.id, "name": k.source, "type": k.source_type } for k in knowledge_items] ) except HTTPException: raise except Exception as e: logger.error(f"Error updating agent groups: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to update agent groups" ) @router.get("/{agent_id}/shopify", response_model=AgentWithCustomizationResponse) @router.get("/{agent_id}", response_model=AgentWithCustomizationResponse) async def get_agent_by_id( agent_id: UUID, auth_info: dict = Depends( require_any_unified_permission("view_agents", "manage_agents")), db: Session = Depends(get_db) ): """Get agent by ID with customization - supports both JWT and Shopify session token auth""" agent_repo = AgentRepository(db) agent = agent_repo.get_by_id(agent_id) if not agent: raise HTTPException(status_code=404, detail="Agent not found") # Check if agent belongs to the authenticated organization verify_agent_access(agent, auth_info) # photo_url is signed by CustomizationResponse on serialization. return agent @router.post("/generate-instructions", response_model=List[str]) @limit_instruction_generation async def generate_instructions( request: Request, prompt_data: InstructionPrompt, current_user: User = Depends(require_permissions("manage_agents")), db: Session = Depends(get_db) ): """Generate agent instructions using AI""" try: # Get prompt and existing instructions from request prompt = prompt_data.prompt.strip() existing_instructions = prompt_data.existing_instructions # Validate prompt if not prompt: raise HTTPException( status_code=400, detail="Prompt cannot be empty" ) if len(prompt) > 1000: raise HTTPException( status_code=400, detail="Prompt exceeds maximum length of 1000 characters" ) # Check if enterprise module is available try: from app.enterprise.repositories.subscription import SubscriptionRepository # noqa: F401 - presence check preserves the optional enterprise contract HAS_ENTERPRISE = True except ImportError: HAS_ENTERPRISE = False # Get API key and model - first check for ReplyPier config if enterprise is available api_key = "" model_name = "" model_type = "" base_url = None if HAS_ENTERPRISE: # Use Groq/OpenAI with keys from env for ReplyPier model model_type = 'OPENAI' model_name = os.getenv('REPLYPIER_MODEL_NAME', 'gpt-4o-mini') api_key = os.getenv('REPLYPIER_API_KEY', '') if not api_key: logger.error("ReplyPier API key not found in environment") raise HTTPException( status_code=500, detail="ReplyPier API configuration missing" ) else: # Get from organization's AI config from app.repositories.ai_config import AIConfigRepository ai_config_repo = AIConfigRepository(db) ai_config = ai_config_repo.get_active_config(current_user.organization_id) if not ai_config: raise HTTPException( status_code=404, detail="No AI configuration found for your organization" ) model_type = ai_config.model_type model_name = ai_config.model_name api_key = ai_config.api_key base_url = (ai_config.settings or {}).get('base_url') if not api_key: raise HTTPException( status_code=500, detail="API configuration missing from your organization settings" ) # Create an instruction generation system message with context system_message = """ You are an expert at creating instructions for AI assistants from scratch or modify existing instructions based on user's prompt. Your task is to generate a set of clear, concise, and effective instructions based on the user's prompt. Follow these guidelines: 1. Each instruction should be specific and actionable 2. Avoid contradictions between instructions 3. Cover different aspects of the agent's behavior and knowledge 4. Include limitations and boundaries where appropriate 5. Format your response as a list of instructions, with each instruction separated by a newline 6. Return between 3-7 instructions total 7. Each instruction should be 1-3 sentences for clarity 8. Focus on creating versatile instructions that will work well for customer service If existing instructions are provided: 1. Review and understand the existing instructions 2. Avoid duplicating existing instructions 3. Focus on complementing or enhancing the existing instructions 4. If modifying existing instructions, preserve their core intent while improving clarity 5. Maintain the existing instructions' order and format when adding new instructions, dont add at starting if existing instructions has starting phrase, rewrite to fit in existing instructions Final output append with existing instruction and give complete one Return ONLY the instructions as a list, with no additional explanation or commentary. The instructions should be immediately usable in an AI agent configuration. """ # Create a temporary agent for generation using the utility function model = create_model( model_type=model_type, api_key=api_key, model_name=model_name, max_tokens=800, base_url=base_url ) generator = AgnoAgent( name="Instruction Generator", model=model, instructions=system_message, debug_mode=settings.ENVIRONMENT == "development" ) # Generate instructions with context user_message = f"Please generate instructions for an AI agent with the following purpose: {prompt}" if existing_instructions: existing_instructions_text = "\n".join(f"- {instr}" for instr in existing_instructions) user_message += f"\n\nExisting instructions for context:\n{existing_instructions_text}" response = await generator.arun(user_message) # Parse response to get instructions list response_text = response.content if hasattr(response, 'content') else str(response) # Split the response into separate instructions instructions = [] for line in response_text.strip().split('\n'): line = line.strip() # Skip empty lines and list markers if line and not line.startswith('```') and not line.endswith('```'): # Remove list markers if present (1., -, *, etc.) clean_line = line if line[0].isdigit() and line[1:3] in ['. ', '- ', ') ']: clean_line = line[3:].strip() elif line[0] in ['-', '*', '•']: clean_line = line[1:].strip() if clean_line: instructions.append(clean_line) # Ensure the instructions are within limits total_length = sum(len(instruction) for instruction in instructions) if total_length > 2000: # Trim instructions to fit within 1500 characters result = [] current_length = 0 for instruction in instructions: if current_length + len(instruction) <= 1500: result.append(instruction) current_length += len(instruction) else: # Try to add a truncated version of the instruction remaining = 2000 - current_length if remaining > 50: # Only add if we can include something meaningful truncated = instruction[:remaining - 3] + "..." result.append(truncated) break instructions = result return instructions except HTTPException: raise except Exception as e: logger.error(f"Error generating instructions: {str(e)}") raise HTTPException( status_code=500, detail="Failed to generate instructions" )