This document defines the standard contract and behavior for all agents in the Purple8 platform. ALL agents MUST follow these guidelines for proper HITL (Human-In-The-Loop) integration.
Every agentβs run() method MUST return a dictionary with the following structure:
{
"human_summary": str, # REQUIRED - Human-readable summary for HITL review
"<primary_output>": Any, # REQUIRED - Main agent output (varies by agent type)
"agent_type": str, # REQUIRED - Agent identifier
"key_decisions": list, # OPTIONAL - List of important decisions made
"metadata": dict, # OPTIONAL - Additional context
}
human_summary RequirementsThe human_summary field is CRITICAL for HITL. It must:
## π¨ Design SummaryExample template:
def _create_human_summary(self, result: str, context: str) -> str:
return f"""## π¨ {self.agent_name} Summary
### What Was Generated
- Brief description of outputs
- Key artifacts created
### Key Decisions
- Decision 1 and rationale
- Decision 2 and rationale
### For HITL Review
β
Checkpoint 1
β
Checkpoint 2
β
Checkpoint 3
"""
Every agent MUST define:
class MyAgent(BaseAgent):
def __init__(self, llm_instance=None):
super().__init__(llm_instance, agent_name="MyAgent")
# Phase and dependencies
self.phase = "planning" # conception|planning|implementation|qa|delivery|safety
self.dependencies = ['previous_agent'] # Agents that must run before this
# Input contract
self.required_inputs = [
{
"name": "requirements",
"type": "string",
"description": "Project requirements from IdeationAgent",
"format": "markdown"
}
]
self.optional_inputs = [
{
"name": "context",
"type": "dict",
"description": "Additional context",
"format": "json"
}
]
# Output contract
self.output_format = "json" # json|markdown|code
self.output_schema = {
"primary_output": {
"type": "string",
"description": "Main output description"
},
"human_summary": {
"type": "string",
"description": "Human-readable summary for HITL"
}
}
ideation β design β architecture β database β api_development
β β β β β
requirements design_spec architecture database_schema api_spec
β
βββββββββββββββββββ¬ββββββββββββββββββ¬ββββββββββββββββββ
β β β β
frontend_development backend_development infrastructure_development
β β β
frontend_code backend_code infrastructure_code
βββββββββββββββββββ΄ββββββββββββββββββ
β
unittest
β
sit
β
uat
β
software_packaging
β
deployment
| From Agent | To Agent | Data Key | Description |
|---|---|---|---|
| ideation | ALL | requirements |
Project requirements specification |
| design | frontend, architecture | design_spec |
UI/UX design document |
| architecture | backend, api, database, infra | architecture |
System architecture |
| database | backend, api, infra | database_schema |
Database design |
| api_development | frontend, backend | api_spec |
OpenAPI specification |
| frontend_development | unittest, packaging | frontend_code |
Frontend source code |
| backend_development | unittest, packaging | backend_code |
Backend source code |
| infrastructure_development | deployment | infrastructure_code |
Docker/K8s configs |
| unittest | sit | unit_test_results |
Unit test suite |
| sit | uat | sit_results |
Integration test results |
| uat | deployment | uat_results |
UAT sign-off |
| software_packaging | deployment | package |
Deployment package |
IdeationAgent - Requirements gatheringDesignAgent - UI/UX designArchitectureAgent - System architectureDatabaseAgent - Database designAPIDevelopmentAgent - API contracts (generates OpenAPI/Swagger)FrontendDevelopmentAgent - Frontend codeBackendDevelopmentAgent - Backend codeInfrastructureDevelopmentAgent - DevOps configsUnitTestAgent - Unit testsSITAgent - System integration testsUATAgent - User acceptance testsSoftwarePackagingAgent - Build configsDeploymentAgent - Deployment plansMobileAgent - Mobile app developmentAIMLAgent - AI/ML featuresRAGAgent - Retrieval-augmented generationSecurityQAAgent - Security testingPerformanceQAAgent - Performance testingObservabilityAgent - Monitoring setupGuardrailsAgent - AI safetyLegalAgent - Compliance checkingDocumentationAgent - Doc generationasync def run(self, requirements: str, upstream_output: Dict = None, **kwargs):
# Extract relevant data from upstream
if upstream_output:
relevant_data = upstream_output.get('key_data', '')
else:
relevant_data = "Default fallback"
def _format_input(self, data: Optional[Dict]) -> str:
if not data:
return "Not provided - use reasonable defaults"
if isinstance(data, str):
return data
return str(data)
async def run(self, ...) -> Dict[str, Any]:
result = await self._invoke_llm(prompt)
# ALWAYS create human summary
human_summary = self._create_human_summary(result)
return {
"primary_output": result,
"human_summary": human_summary,
"agent_type": "my_agent"
}
import logging
logger = logging.getLogger(__name__)
class MyAgent(BaseAgent):
async def run(self, ...):
logger.info(f"{self.agent_name} starting...")
# ... work ...
logger.info(f"{self.agent_name} completed")
BaseAgentphase and dependenciesrequired_inputs and optional_inputsoutput_format and output_schemahuman_summary fieldshared/agents/registry.pyFOUNDATIONAL_AGENTS or specialized list in orchestratorservices/gateway/routers/pipeline.pyThe APIDevelopmentAgent generates:
The Swagger/OpenAPI spec can be:
/docsopenapi.json or openapi.yaml