feat(agent): add Task Agent for structured task execution and improve context management

Author: Davidson Gomes

CHANGELOG.md

@@ -10,6 +10,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - Add CrewAI agents
+- Add Task Agent for structured single-task execution
+- Improve context management in agent execution
 
 ## [0.0.9] - 2025-05-13
 

README.md

@@ -187,6 +187,39 @@ Allows organizing agents into a "crew" with specific tasks assigned to each agen
 }
 ```
 
+### 8. Task Agent
+
+Executes a specific task using a target agent. Task Agent provides a streamlined approach for structured task execution, where the agent_id specifies which agent will process the task, and the task description can include dynamic content placeholders.
+
+```json
+{
+  "client_id": "{{client_id}}",
+  "name": "web_search_task",
+  "type": "task",
+  "folder_id": "folder_id (optional)",
+  "config": {
+    "tasks": [
+      {
+        "agent_id": "search-agent-uuid",
+        "description": "Search the web for information about {content}",
+        "expected_output": "Comprehensive search results with relevant information"
+      }
+    ],
+    "sub_agents": ["post-processing-agent-uuid"]
+  }
+}
+```
+
+Key features of Task Agent:
+
+- Passes structured task instructions to the designated agent
+- Supports variable content using {content} placeholder in the task description
+- Provides clear task definition with instructions and expected output format
+- Can execute sub-agents after the main task is completed
+- Simplifies orchestration for single-focused task execution
+
+Task Agent is ideal for scenarios where you need to execute a specific, well-defined task with clear instructions and expectations.
+
 ### Common Characteristics
 
 - All agent types can have sub-agents

migrations/versions/2df073c7b564_add_task_agent_type_agents_table.py

@@ -0,0 +1,43 @@
+"""add_task_agent_type_agents_table
+
+Revision ID: 2df073c7b564
+Revises: 611d84e70bb2
+Create Date: 2025-05-14 11:46:39.573247
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "2df073c7b564"
+down_revision: Union[str, None] = "611d84e70bb2"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint("check_agent_type", "agents", type_="check")
+    op.create_check_constraint(
+        "check_agent_type",
+        "agents",
+        "type IN ('llm', 'sequential', 'parallel', 'loop', 'a2a', 'workflow', 'crew_ai', 'task')",
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint("check_agent_type", "agents", type_="check")
+    op.create_check_constraint(
+        "check_agent_type",
+        "agents",
+        "type IN ('llm', 'sequential', 'parallel', 'loop', 'a2a', 'workflow', 'crew_ai')",
+    )
+    # ### end Alembic commands ###

src/models/models.py

@@ -123,7 +123,7 @@ class Agent(Base):
 
     __table_args__ = (
         CheckConstraint(
-            "type IN ('llm', 'sequential', 'parallel', 'loop', 'a2a', 'workflow', 'crew_ai')",
+            "type IN ('llm', 'sequential', 'parallel', 'loop', 'a2a', 'workflow', 'crew_ai', 'task')",
             name="check_agent_type",
         ),
     )

src/schemas/schemas.py

@@ -98,7 +98,7 @@ class AgentBase(BaseModel):
     goal: Optional[str] = Field(None, description="Agent goal or objective")
     type: str = Field(
         ...,
-        description="Agent type (llm, sequential, parallel, loop, a2a, workflow, crew_ai)",
+        description="Agent type (llm, sequential, parallel, loop, a2a, workflow, crew_ai, task)",
     )
     model: Optional[str] = Field(
         None, description="Agent model (required only for llm type)"
@@ -137,9 +137,10 @@ def validate_type(cls, v):
             "a2a",
             "workflow",
             "crew_ai",
+            "task",
         ]:
             raise ValueError(
-                "Invalid agent type. Must be: llm, sequential, parallel, loop, a2a, workflow or crew_ai"
+                "Invalid agent type. Must be: llm, sequential, parallel, loop, a2a, workflow, crew_ai or task"
             )
         return v
 
@@ -199,7 +200,7 @@ def validate_config(cls, v, values):
                 raise ValueError(
                     f'Agent {values["type"]} must have at least one sub-agent'
                 )
-        elif values["type"] == "crew_ai":
+        elif values["type"] == "crew_ai" or values["type"] == "task":
             if not isinstance(v, dict):
                 raise ValueError(f'Invalid configuration for agent {values["type"]}')
             if "tasks" not in v:

src/services/agent_builder.py

@@ -41,6 +41,7 @@
 from src.services.custom_agents.a2a_agent import A2ACustomAgent
 from src.services.custom_agents.workflow_agent import WorkflowAgent
 from src.services.custom_agents.crew_ai_agent import CrewAIAgent
+from src.services.custom_agents.task_agent import TaskAgent
 from src.services.apikey_service import get_decrypted_api_key
 from sqlalchemy.orm import Session
 from contextlib import AsyncExitStack
@@ -314,6 +315,55 @@ async def build_workflow_agent(
             logger.error(f"Error building Workflow agent: {str(e)}")
             raise ValueError(f"Error building Workflow agent: {str(e)}")
 
+    async def build_task_agent(
+        self, root_agent
+    ) -> Tuple[TaskAgent, Optional[AsyncExitStack]]:
+        """Build a task agent with its sub-agents."""
+        logger.info(f"Creating Task agent: {root_agent.name}")
+
+        agent_config = root_agent.config or {}
+
+        if not agent_config.get("tasks"):
+            raise ValueError("tasks are required for Task agents")
+
+        try:
+            # Get sub-agents if there are any
+            sub_agents = []
+            if root_agent.config.get("sub_agents"):
+                sub_agents_with_stacks = await self._get_sub_agents(
+                    root_agent.config.get("sub_agents")
+                )
+                sub_agents = [agent for agent, _ in sub_agents_with_stacks]
+
+            # Additional configurations
+            config = root_agent.config or {}
+
+            # Convert tasks to the expected format by TaskAgent
+            tasks = []
+            for task_config in config.get("tasks", []):
+                task = CrewAITask(
+                    agent_id=task_config.get("agent_id"),
+                    description=task_config.get("description", ""),
+                    expected_output=task_config.get("expected_output", ""),
+                )
+                tasks.append(task)
+
+            # Create the Task agent
+            task_agent = TaskAgent(
+                name=root_agent.name,
+                tasks=tasks,
+                db=self.db,
+                sub_agents=sub_agents,
+            )
+
+            logger.info(f"Task agent created successfully: {root_agent.name}")
+
+            return task_agent, None
+
+        except Exception as e:
+            logger.error(f"Error building Task agent: {str(e)}")
+            raise ValueError(f"Error building CrewAI agent: {str(e)}")
+
     async def build_crew_ai_agent(
         self, root_agent: Agent
     ) -> Tuple[CrewAIAgent, Optional[AsyncExitStack]]:
@@ -434,7 +484,8 @@ async def build_agent(self, root_agent) -> Tuple[
         | LoopAgent
         | A2ACustomAgent
         | WorkflowAgent
-        | CrewAIAgent,
+        | CrewAIAgent
+        | TaskAgent,
         Optional[AsyncExitStack],
     ]:
         """Build the appropriate agent based on the type of the root agent."""
@@ -446,5 +497,7 @@ async def build_agent(self, root_agent) -> Tuple[
             return await self.build_workflow_agent(root_agent)
         elif root_agent.type == "crew_ai":
             return await self.build_crew_ai_agent(root_agent)
+        elif root_agent.type == "task":
+            return await self.build_task_agent(root_agent)
         else:
             return await self.build_composite_agent(root_agent)

src/services/agent_service.py

@@ -202,7 +202,7 @@ async def create_agent(db: Session, agent: AgentCreate) -> Agent:
             if "api_key" not in agent.config or not agent.config["api_key"]:
                 agent.config["api_key"] = generate_api_key()
 
-        elif agent.type == "crew_ai":
+        elif agent.type == "crew_ai" or agent.type == "task":
             if not isinstance(agent.config, dict):
                 agent.config = {}
                 raise HTTPException(
@@ -213,7 +213,7 @@ async def create_agent(db: Session, agent: AgentCreate) -> Agent:
             if "tasks" not in agent.config:
                 raise HTTPException(
                     status_code=status.HTTP_400_BAD_REQUEST,
-                    detail="Invalid configuration: tasks is required for crew_ai agents",
+                    detail=f"Invalid configuration: tasks is required for {agent.type} agents",
                 )
 
             if not agent.config["tasks"]:
@@ -682,14 +682,14 @@ async def update_agent(
         if "config" not in agent_data:
             agent_data["config"] = agent_config
 
-        if ("type" in agent_data and agent_data["type"] == "crew_ai") or (
-            agent.type == "crew_ai" and "config" in agent_data
+        if ("type" in agent_data and agent_data["type"] in ["crew_ai", "task"]) or (
+            agent.type in ["crew_ai", "task"] and "config" in agent_data
         ):
             config = agent_data.get("config", {})
             if "tasks" not in config:
                 raise HTTPException(
                     status_code=status.HTTP_400_BAD_REQUEST,
-                    detail="Invalid configuration: tasks is required for crew_ai agents",
+                    detail=f"Invalid configuration: tasks is required for {agent_data.get('type', agent.type)} agents",
                 )
 
             if not config["tasks"]:

src/services/custom_agents/crew_ai_agent.py

@@ -1,9 +1,9 @@
 """
 ┌──────────────────────────────────────────────────────────────────────────────┐
 │ @author: Davidson Gomes                                                      │
-│ @file: a2a_agent.py                                                          │
+│ @file: crew_ai_agent.py                                                      │
 │ Developed by: Davidson Gomes                                                 │
-│ Creation date: May 13, 2025                                                  │
+│ Creation date: May 14, 2025                                                  │
 │ Contact: [email protected]                                           │
 ├──────────────────────────────────────────────────────────────────────────────┤
 │ @copyright © Evolution API 2025. All rights reserved.                        │