triad-core-ai 0.1.0

A lightweight AI Agent orchestration framework with unified Task/Prompt/Function management and parallel routing

Triad Core AI Framework Technical Manual

Version: v1.1
Use Case: Enterprise-grade AI Agent Development, Production Deployment
Core Features: Three-tier Memory Isolation, Dynamic Runtime Registration, Native MCP Support, DAG Task Orchestration

---

1. Product Overview

1.1 Introduction

Triad Core AI is a Python framework for building structured, multi-step AI Agent workflows. It provides a Task-based orchestration engine that lets you compose complex AI interactions into executable graphs made of composable, reusable tasks — each with its own Prompt, tool functions, conversation context, and data flow.

The name "Triad" comes from the three core concepts managed through the central Registry: Task, Prompt, and Function, which together form the capability foundation of the framework.

1.2 Core Triad

Concept	Description
Task	An independent unit encapsulating business logic and AI interactions; the smallest granularity of orchestration
Prompt	Jinja2-templated prompts with automatic variable extraction; 1:1 binding with Task
Function	Tools callable by AI, with parameters defined using JSON Schema; supports local functions and MCP remote tools

1.3 Feature Highlights

• 🧩 Declarative Task Definition — Define AI tasks with concise decorators (@task, @prompt, @function), lowering the barrier to entry
• 🔗 Task Chain Routing — Tasks can route to one or more downstream tasks via next_task, forming a DAG execution graph
• ⚡ Parallel Execution — Independent tasks in the same layer run concurrently via asyncio, improving throughput
• 🛠 Automatic Tool Loop — Built-in call_ai_e2e() handles multi-turn tool call loops automatically, no manual dialog state management needed
• 📦 Multi-layer Context — Supports session-level (cross-request), task-level (per-task private), and request-level (single-call temporary) data isolation
• 💬 Intelligent Conversation History — Task-isolated conversation history (auto-cleaned after task ends) + global semantic task history (cross-task long-term memory)
• 🔄 Pluggable AI Providers — Built-in OpenAI provider (compatible with OpenAI, DeepSeek, Azure, etc.); custom extensions supported
• ⏱ High Availability — Automatic retries with exponential backoff, configurable timeouts, suitable for unstable network environments
• 💾 Session Persistence — Supports in-memory storage (development) and Redis storage (production, including standalone/sentinel/cluster modes)
• 📋 Comprehensive Error Handling — Full exception hierarchy for different failure modes, facilitating issue diagnosis and circuit-breaking

---

2. Core Concepts

This section dives into the details of the core triad and the surrounding concepts that power the framework:

2.1 Task

A Task is the encapsulation unit for business logic, divided into two categories:

• Base Task: Inherits from the BaseTask abstract class, suitable for deterministic business logic
• AI Task: Inherits from the AITask base class, with built-in AI invocation, tool loop, and context management capabilities, suitable for LLM-driven flexible logic

Each task has a unique task_id, and conversation history is automatically isolated during execution to prevent cross-task context contamination. Tasks declare their dependent tools via required_functions, which are automatically validated for existence at registration time.

2.2 Prompt

A Prompt is the AI's instruction template, using Jinja2 syntax with support for variable interpolation. The framework automatically extracts variables from the template and injects values at render time, in the following priority order:

1. Parameters explicitly passed to the render() method (highest priority)
2. Current task's private data (Context.get())
3. Global session data (Context.get_session())

StrictUndefined Mode Enabled: If a variable is undefined, a RuntimeError is raised directly at render time, preventing the AI from receiving incorrect information.

2.3 Function

A Function is an atomic tool callable by AI, divided into two categories:

• Local Function: Python synchronous/asynchronous functions, bound via the @function decorator or dynamic registration
• MCP Tool: Remote tools accessed via the MCP protocol; the framework automatically converts them into internal Functions with the naming format server_name.tool_name

Function parameters are defined using JSON Schema, and the AI automatically generates valid call parameters based on the Schema.

2.4 Supporting Core Concepts

Concept	Scope	Description
Context	Request-level	State container for a single request, managing multi-layer data and conversation history
Registry	Global singleton	Central registry storing all Task, Prompt, and Function definitions; supports static and dynamic registration
Runtime	Request-level	DAG orchestration engine responsible for task scheduling, parallel execution, and chain progression
Agent	Task-level	Single-task executor handling retries, timeout control, and exception capture
Session	User-level	Cross-request persistent storage for user state and long-term preferences

---

3. Project Structure

The framework adopts a modular design with a clear directory structure, facilitating secondary development and troubleshooting:

triad-core-ai/
├── src/
│   └── triad_core_ai/                  # Framework core source code
│       ├── __init__.py                 # Public API exports
│       ├── ai_task.py                  # AITask base class: AI-driven tasks with built-in tool loop logic
│       ├── exceptions.py               # Global exception hierarchy definition
│       ├── schemas.py                  # Pydantic data models: TaskInput, TaskOutput, etc.
│       ├── task.py                     # BaseTask abstract class: parent class for all tasks
│       ├── core/                       # Core runtime modules
│       │   ├── agent.py                # Agent executor + AgentPool unified entry point
│       │   ├── context.py              # Multi-layer context container implementation
│       │   ├── runtime.py              # DAG task orchestration engine
│       │   └── session.py              # Session persistence implementation (in-memory/Redis)
│       ├── interfaces/                 # Abstract interface definitions
│       │   └── ai_provider.py          # AIProvider abstract base class, ToolCall, ChatResponse definitions
│       ├── providers/                  # AI provider implementations
│       │   ├── mock_provider.py        # Mock AI provider (for unit testing)
│       │   └── openai_provider.py      # OpenAI-compatible provider (supports OpenAI/DeepSeek/Azure/etc.)
│       └── registry/                   # Registry module (static + dynamic registration support)
│           ├── decorators.py           # Static decorator implementations
│           └── registry.py             # Global Registry singleton + dynamic registration implementation
├── pyproject.toml                      # Project configuration file (dependencies, packaging, lint rules)
├── README.md                           # English documentation
├── README_CN.md                        # Chinese documentation
└── main.py                             # Example entry point

# Optional extension modules (import on demand)
└── triad_core_ai/
    └── mcp/                            # MCP protocol support extension
        ├── adapter.py                  # MCP Server adapter
        ├── handler.py                  # Unified MCP tool execution entry point
        ├── registry.py                 # MCP registry center
        └── schema.py                   # MCP server configuration model

---

4. Registration Mechanism

⚠️ Important: Dynamic registration is NOT exclusive to AI-generated scenarios. It is a "runtime programmable" capability suitable for admin dashboards, plugin architectures, hot-reload configurations, and any scenario requiring capability registration without service restart.

Static and dynamic registrations both write to the same global Registry and are fully interoperable.

4.1 Static Registration

Located in src/triad_core_ai/registry/decorators.py, suitable for hardened business logic with optimal performance.

Note: All statically registered functions, prompts, and tasks need to be imported in __init__ to be recognized by the runtime.

@function — Registering Tool Functions

from triad_core_ai.registry.decorators import function

@function(
    name="query_order",               # Unique tool identifier; for MCP tools, use format: server_name.tool_name
    description="Query order details by order number",  # AI-readable tool description; clearer = higher accuracy
    parameters={                      # JSON Schema format, defining input constraints
        "type": "object",
        "properties": {
            "order_id": {
                "type": "string",
                "description": "Order number, format: ORDxxxxxx"
            },
            "include_items": {
                "type": "boolean",
                "default": False,
                "description": "Whether to return order item details"
            }
        },
        "required": ["order_id"]       # Required parameter list
    }
)
def query_order(order_id: str, include_items: bool = False) -> dict:
    """Business logic implementation; supports sync/async functions"""
    # Database query logic
    return {"order_id": order_id, "status": "shipped"}

@prompt — Registering System Prompts

from triad_core_ai.registry.decorators import prompt

@prompt(
    name="customer_service_role",      # Unique prompt identifier; defaults to Task name
    description="E-commerce customer service role setting"  # Prompt description
)
def cs_prompt():
    """Returns a Jinja2 template string; variables are auto-injected from Context"""
    return """
    You are an e-commerce customer service assistant. Current user ID: {{ user_id }}
    Recent inquiry history: {{ recent_queries }}
    Please reply in strict JSON format with fields: answer, need_transfer
    """

• Template variable injection priority: render() explicit params > Context.task_data > Context.session_data
• StrictUndefined enabled: Undefined variables raise errors immediately, preventing AI from receiving bad information

@task — Registering Business Tasks

from triad_core_ai.registry.decorators import task
from triad_core_ai.ai_task import AITask
from triad_core_ai.schemas import TaskInput, TaskOutput

@task(
    name="after_sale_process",         # Unique task identifier
    description="After-sales issue handling workflow",  # Task description
    required_functions=[               # Tools this task depends on; auto-validated at registration
        "query_order",
        "refund_order"
    ]
)
class AfterSaleTask(AITask):
    async def execute(self, inp: TaskInput) -> TaskOutput:
        # 1. Prepare context variables
        self.context.set("recent_queries", inp.user_message)
        
        # 2. Call AI (automatically handles Tool Loop)
        response = await self.call_ai_e2e(
            max_tool_loops=3,          # Maximum tool call rounds
            max_lazy_retries=2         # AI empty-reply retry count
        )
        
        # 3. Return result
        return TaskOutput(
            success=True,
            ai_response=response.content,
            data={"refund_id": self.context.get("refund_id")},
            next_task=["notify_user"] if response.content else None,  # Downstream tasks
            user_visible=True,         # Result visible to user
            ai_visible=True            # Result visible to AI (written to global memory)
        )

4.2 Dynamic Registration

Located in src/triad_core_ai/registry/, suitable for admin configurations, plugin architectures, and hot-reload scenarios — add capabilities without modifying code.

DynamicFunctionRegistry — Dynamically Register Functions

Used to wrap existing business functions as AI tools, or to add new tool definitions at runtime.

from triad_core_ai.registry import get_dynamic_function_registry

# 1. Get the global dynamic function registry
dyn_func_reg = get_dynamic_function_registry()

# 2. Bind an existing business function (most common usage)
def legacy_refund_logic(order_id: str, amount: float):
    """Existing business function; no need to modify original code"""
    return {"refund_id": "RFxxxxxx", "status": "success"}

dyn_func_reg.register(
    name="legacy_refund",              # Tool name
    handler=legacy_refund_logic,       # Bind existing execution function
    description="Interface to legacy refund system",
    parameters={
        "type": "object",
        "properties": {
            "order_id": {"type": "string"},
            "amount": {"type": "number"}
        },
        "required": ["order_id", "amount"]
    }
)

# 3. Register definition only (no custom handler, uses default placeholder)
# Suitable for scenarios where the handler will be bound later via other mechanisms
dyn_func_reg.register(
    name="temp_tool",
    handler=lambda **kwargs: f"Temp tool called: {kwargs}",
    description="Temporary tool with default handler",
    parameters={"type": "object", "properties": {"x": {"type": "string"}}}
)

DynamicTaskRegistry — Dynamically Register Tasks

Used to add tasks at runtime; the framework automatically performs prompt validation and Task class generation.

from triad_core_ai.registry import get_dynamic_task_registry
from triad_core_ai.dynamic.schemas import DynamicTaskDefinition, DynamicFunctionSpec

# 1. Get the global dynamic task registry
dyn_task_reg = get_dynamic_task_registry()

# 2. Define the task (essentially a Pydantic model, can be parsed directly from API request body)
task_def = DynamicTaskDefinition(
    task_name="dynamic_after_sale",    # Task name, should match Prompt name
    prompt_template="""User question: {{ user_question }}
    Please call query_order to check order status, then decide whether to refund.""",  # Jinja2 template
    required_functions=["query_order"], # Dependencies on already-registered functions
    new_functions=[                     # New function definitions (optional)
        DynamicFunctionSpec(
            name="query_order",
            description="Query order",
            parameters={"properties": {"order_id": {"type": "string"}}}
        )
    ],
    timeout=60,                         # Task timeout in seconds
    max_tool_loops=3,                   # Maximum tool call rounds
    max_lazy_retries=2                  # AI empty-reply retry count
)

# 3. Register (automatically validates Prompt variables against function parameters)
registered_name = dyn_task_reg.register(task_def)
print(f"Dynamic task registered successfully: {registered_name}")

• Automatic validation at registration: Checks whether variables referenced in the Prompt exist in the parameters of required_functions
• Automatic generation: The framework automatically creates a dynamic Task class inheriting from AITask, no manual class definition needed

---

5. Session Storage Configuration

Session storage is responsible for persisting session_data (user login state, history preferences, business state, etc.) across requests. Both in-memory and Redis backends are supported.

5.1 In-Memory Storage (Development Environment)

Default configuration; data is stored in the Python process memory with no additional dependencies required.

from triad_core_ai.core.session import InMemorySessionStore

session_store = InMemorySessionStore()
# No additional parameters; suitable for single-process development and debugging
# Drawbacks: Data lost on service restart; not shared across multi-process/multi-instance deployments

5.2 Redis Storage (Production Environment, Recommended)

Located in src/triad_core_ai/core/session.py, supports standalone, sentinel, and cluster modes. Install the dependency first:

pip install redis>=4.0

Basic Configuration (Standalone Redis)

from triad_core_ai.core.session import RedisSessionStore

session_store = RedisSessionStore(
    url="redis://:password@localhost:6379/0",  # Redis connection string
    prefix="triad:session:",                    # Key prefix for multi-app isolation
    ttl=3600                                    # Session expiration time in seconds (default: 1 hour)
)

Production-Grade Configuration (Custom Connection Pool)

import redis.asyncio as aioredis
from triad_core_ai.core.session import RedisSessionStore

# 1. Custom Redis client (supports connection pooling, SSL, timeouts, etc.)
redis_client = aioredis.Redis(
    host="redis-host",
    port=6379,
    password="password",
    db=0,
    decode_responses=True,       # Auto-decode to strings; no manual bytes-to-str conversion
    max_connections=100,         # Connection pool size; adjust based on QPS
    socket_timeout=5,            # Socket timeout
    socket_connect_timeout=5,     # Connection timeout
    health_check_interval=30     # Connection health check interval
)

# 2. Inject custom client
session_store = RedisSessionStore(
    url="redis://ignored",       # URL is ignored when a custom client is injected
    prefix="triad:prod:session:",
    ttl=7200                     # Production environments should use TTL >= 2 hours
)
session_store._client = redis_client

Sentinel Mode Configuration

from redis.asyncio.sentinel import Sentinel
from triad_core_ai.core.session import RedisSessionStore

# 1. Configure sentinel
sentinel = Sentinel(
    sentinels=[("sentinel-1", 26379), ("sentinel-2", 26379)],
    password="sentinel-password",
    socket_timeout=5
)

# 2. Get master node client
master = sentinel.master_for(
    service_name="mymaster",
    password="redis-password",
    decode_responses=True,
    max_connections=100
)

# 3. Inject client
session_store = RedisSessionStore(url="redis://ignored")
session_store._client = master

Cluster Mode Configuration

from redis.asyncio.cluster import RedisCluster
from triad_core_ai.core.session import RedisSessionStore

# 1. Configure cluster
redis_cluster = RedisCluster(
    host="redis-cluster-host",
    port=7000,
    password="cluster-password",
    decode_responses=True,
    max_connections=100
)

# 2. Inject client
session_store = RedisSessionStore(url="redis://ignored")
session_store._client = redis_cluster

Integrating Session Storage in AgentPool

from triad_core_ai.core.agent_pool import AgentPool

pool = AgentPool(
    session_id="user_12345",       # Unique user identifier, used as Redis key suffix
    ai_provider=ai_provider,       # AI provider instance
    registry=registry,             # Global registry
    session_store=session_store,   # Redis session store instance
    session_data={},               # Initial data (overwritten if exists in Redis)
    mcp_servers=[]                 # MCP server configurations (optional)
)

# After each process() call, the framework automatically saves Context.session_data to Redis
result = await pool.process(
    user_message="Check my orders",
    initial_task="after_sale_process"
)

---

6. Orchestration Mechanism

6.1 Core Dependency Relationships

These are strongly bound relationships, with dependency validity automatically verified at registration:

• Task → depends on → Prompt (1:1 binding; default: Task name = Prompt name)
• Task → depends on → Function (1:N binding via required_functions)
• Prompt → references → Context variables (no strong dependency; runtime-checked)

6.2 Complete Execution Flow

┌─ User Request ─→ AgentPool.process()
    │
    ├─ 1. Load Session: Load session_data from SessionStore into Context
    │
    ├─ 2. Initialize Runtime: Create Context, generate runtime_task_id, write user message
    │
    ├─ 3. DAG Scheduling: current_layer = [initial_task]
    │
    ├─ 4. Parallel Execute Current Layer: asyncio.gather(*tasks)
    │   │
    │   ├─ Agent.run(): Handle retries/timeouts
    │   │   │
    │   │   └─ AITask.execute(): Business logic entry point
    │   │       │
    │   │       ├─ Build Prompt: Inject variables from Context
    │   │       │
    │   │       ├─ Assemble Messages:
    │   │       │   ├─ System Prompt (Task-bound Prompt)
    │   │       │   ├─ Task History (global semantic memory, ai_visible=True)
    │   │       │   ├─ Conversation History (current Task isolated)
    │   │       │   └─ User Message
    │   │       │
    │   │       ├─ call_ai_e2e(): AI invocation + Tool Loop
    │   │       │   ├─ Call AI API
    │   │       │   ├─ If tool_calls exist: Execute functions (local/MCP)
    │   │       │   ├─ Write tool results to Conversation History
    │   │       │   └─ Repeat until AI returns final text
    │   │       │
    │   │       └─ Archive: Write result to Task History
    │   │
    │   └─ Collect TaskOutput, update session_data
    │
    ├─ 5. Chain Progression: If TaskOutput.next_task is non-empty, add to next layer's execution queue
    │
    └─ 6. Persist: Save Context.session_data back to SessionStore, return result

6.3 Chained Task Scheduling

Task relay is implemented via TaskOutput.next_task, supporting serial, parallel, and conditional branching:

class OrderQueryTask(AITask):
    async def execute(self, inp: TaskInput) -> TaskOutput:
        order_id = inp.user_message.split()[-1]
        order_info = await query_order(order_id)
        # Store order status in session for downstream tasks
        self.context.set_session("order_status", order_info["status"])
        
        # Determine downstream tasks based on status
        if order_info["status"] == "shipped":
            next_tasks = ["logistics_query"]
        elif order_info["status"] == "pending":
            next_tasks = ["remind_pay"]
        else:
            next_tasks = []
            
        return TaskOutput(
            success=True,
            data={"order_info": order_info},
            next_task=next_tasks
        )

• Best practice for data passing: Cross-task data should be shared via Context.set_session() / get_session(), not previous_output, to avoid data races in parallel scenarios
• Parallel execution: Multiple tasks in the same layer execute concurrently, ideal for independent tasks (e.g., querying orders and user info simultaneously)

6.4 Memory Isolation Model

The framework achieves four-tier memory isolation to彻底 eliminate cross-task context contamination:

┌─ Session Store (Redis) ─────────────────────┐
│  session_data: {user_id:123, token:"xxx"}   │  # Globally shared, cross-request persistent
└─────────────────────┬───────────────────────┘
                      ↓ Load/Save
┌─ Context (Single Request) ──────────────────┐
│  ├─ session_data: Copy from Session Store   │  # Globally shared, readable/writable within request
│  ├─ task_data: Temporary vars, destroyed after Task ends │  # Task-private
│  ├─ task_history: [...]                    │  # Global semantic memory, shared across Tasks
│  └─ conversation_histories: {               │  # Task-isolated conversation history
│       "task_1": [sys, user, asst, tool],    │  # Visible only to Task 1
│       "task_2": [sys, user, asst],          │  # Visible only to Task 2
│       "runtime_task_id": [...]              │  # Root Runtime task history
│     }
└─────────────────────────────────────────────┘

• Conversation History: Strictly isolated by task_id, ensuring multi-turn conversation contexts don't interfere with each other
• Task History: Globally shared semantic summary; only stores results with ai_visible=True, used to inject long-term memory into AI
• Auto-cleanup: Conversation History for a Task is automatically cleaned up after the Task ends, preventing memory leaks

---

7. API Quick Reference

Core Class Initialization Parameters

Agent

Parameter	Type	Default	Description
task_def	TaskDefinition	Required	Task definition
ai_provider	AIProvider	Required	AI provider instance
context	Context	Required	Context instance
registry	Registry	Required	Global registry
max_retries	int	3	Retry count for timeouts/network errors
timeout	int	30	Single-task timeout in seconds

Runtime

Parameter	Type	Default	Description
user_message	str	Required	Original user input
registry	Registry	Required	Global registry
ai_provider	AIProvider	Required	AI provider instance
session_data	dict	Required	Session data
task	str	Required	Root task name
max_iterations	int	50	Maximum task chain depth
wait_timeout	int	120	Task wait timeout in seconds

AgentPool

Parameter	Type	Default	Description
session_id	str	Required	Unique user identifier
ai_provider	AIProvider	Required	AI provider instance
registry	Registry	Required	Global registry
session_store	SessionStore	None	Session store instance
session_data	dict	{}	Initial session data
mcp_servers	List[dict]	[]	MCP server configuration list

OpenAIProvider

Parameter	Type	Default	Description
api_key	str	Required	API key
base_url	str	Official OpenAI URL	Compatible endpoint (DeepSeek/Azure/Ollama, etc.)
model	str	gpt-4o-mini	Model name
enable_json_mode	bool	True	Whether to enable JSON output mode
timeout	float	60.0	Request timeout in seconds
max_retries	int	3	API call retry count

Common Context Methods

Method	Parameters	Return	Description
set(key, value)	key: str, value: Any	None	Set Task-level temporary variable
get(key)	key: str	Any	Get Task-level temporary variable
set_session(key, value)	key: str, value: Any	None	Set Session-level persistent variable
get_session(key)	key: str	Any	Get Session-level persistent variable
add_conversation_entry(...)	role, content, task_id, etc.	None	Write to specified Task's conversation history
add_task_history_entry(...)	task_name, content, user_visible, ai_visible	None	Write to global semantic memory
get_recent_task_history(n=3)	n: int	List[dict]	Get the last n entries from global semantic memory

---

8. Practical Examples

Example 1: Standard Static Registration (Full Workflow)

import asyncio
from triad_core_ai.registry.decorators import function, prompt, task
from triad_core_ai.ai_task import AITask
from triad_core_ai.schemas import TaskInput, TaskOutput
from triad_core_ai.providers.openai_provider import OpenAIProvider
from triad_core_ai.core.runtime import Runtime
from triad_core_ai.core.session import InMemorySessionStore

# 1. Define the tool
@function(
    name="calc_sum",
    description="Calculate the sum of two numbers",
    parameters={"properties": {"a": {"type": "number"}, "b": {"type": "number"}}, "required": ["a", "b"]}
)
def calc_sum(a: float, b: float) -> float:
    return a + b

# 2. Define the Prompt
@prompt(name="calc_role")
def calc_prompt():
    return "You are a calculator. Call the calc_sum tool to compute the sum of two numbers provided by the user. Return the result in JSON format."

# 3. Define the Task
@task(name="calc_task", required_functions=["calc_sum"])
class CalcTask(AITask):
    async def execute(self, inp: TaskInput) -> TaskOutput:
        response = await self.call_ai_e2e()
        return TaskOutput(success=True, ai_response=response.content, user_visible=True)

# 4. Run
async def main():
    ai_provider = OpenAIProvider(api_key="sk-xxx")
    registry = get_registry()  # Global registry singleton
    session_store = InMemorySessionStore()
    
    runtime = Runtime(
        user_message="Calculate 3.5 plus 4.2",
        registry=registry,
        ai_provider=ai_provider,
        session_data={},
        task="calc_task"
    )
    
    result = await runtime.run()
    print(result["outputs"]["calc_task"]["ai_response"])

if __name__ == "__main__":
    asyncio.run(main())

Example 2: Dynamic Registration (Admin Config Scenario, Not AI-Generated)

import asyncio
from triad_core_ai.registry import get_dynamic_task_registry
from triad_core_ai.dynamic.schemas import DynamicTaskDefinition
from triad_core_ai.providers.openai_provider import OpenAIProvider
from triad_core_ai.core.runtime import Runtime
from triad_core_ai.core.session import RedisSessionStore
from triad_core_ai.registry.registry import get_registry

# Simulate loading config from database
async def load_task_config_from_db():
    return {
        "task_name": "db_configured_task",
        "prompt_template": "User question: {{ question }}. Please call knowledge_base to find the answer.",
        "required_functions": ["knowledge_base"],
        "timeout": 45
    }

async def main():
    # 1. Initialize core components
    ai_provider = OpenAIProvider(api_key="sk-xxx")
    registry = get_registry()
    session_store = RedisSessionStore(url="redis://localhost:6379/0")
    
    # 2. Load config from database and dynamically register task (no service restart needed)
    config = await load_task_config_from_db()
    dyn_task_reg = get_dynamic_task_registry()
    dyn_task_reg.register(DynamicTaskDefinition(**config))
    print("Dynamic task registration complete")
    
    # 3. Run the dynamically registered task
    runtime = Runtime(
        user_message="How do I return an item?",
        registry=registry,
        ai_provider=ai_provider,
        session_data={},
        task="db_configured_task"
    )
    
    result = await runtime.run()
    print(result["outputs"]["db_configured_task"]["ai_response"])

if __name__ == "__main__":
    asyncio.run(main())

Example 3: MCP Tool Integration

import asyncio
from triad_core_ai.mcp.schema import MCPServerConfig
from triad_core_ai.mcp.registry import MCPRegistry
from triad_core_ai.providers.openai_provider import OpenAIProvider
from triad_core_ai.core.agent_pool import AgentPool
from triad_core_ai.core.session import RedisSessionStore
from triad_core_ai.registry.registry import get_registry

async def main():
    # 1. Initialize core components
    ai_provider = OpenAIProvider(api_key="sk-xxx")
    registry = get_registry()
    session_store = RedisSessionStore(url="redis://localhost:6379/0")
    
    # 2. Configure MCP server (official filesystem server)
    mcp_config = MCPServerConfig(
        name="filesystem",
        transport="stdio",
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
        enabled=True
    )
    
    # 3. Register MCP server (automatically converts MCP tools to internal Functions, naming: filesystem.read_file)
    mcp_registry = MCPRegistry()
    await mcp_registry.register_server(mcp_config)
    
    # 4. Initialize AgentPool
    pool = AgentPool(
        session_id="user_123",
        ai_provider=ai_provider,
        registry=registry,
        session_store=session_store,
        mcp_servers=[mcp_config.model_dump()]
    )
    
    # 5. Execute task (assumes a task exists that calls filesystem tools)
    result = await pool.process(
        user_message="Read the contents of /tmp/test.txt",
        initial_task="file_read_task"
    )
    print(result)

if __name__ == "__main__":
    asyncio.run(main())

---

9. Best Practices

1. Registration mode selection:

   • Use static registration for core business logic and high-frequency capabilities — better performance and type safety
   • Use dynamic registration for admin configurations, plugin architectures, and hot-reload scenarios

2. Session storage guidelines:

   • Production environments MUST use Redis; set TTL based on business needs (recommend >= 2 hours for customer service scenarios)
   • Use consistent key prefix conventions to avoid conflicts across applications

3. Prompt writing guidelines:

   • Clearly define the AI's role, task, and output format
   • Split complex logic into multiple Tasks; avoid overly long single Prompts
   • When JSON mode is enabled, explicitly require the AI to return JSON conforming to the Schema

4. Error handling:

   • On task execution failure, return success=False with a clear error message
   • Tool call failures should return structured error information to help the AI understand the failure reason

5. Performance optimization:

   • Minimize unnecessary Task History entries to avoid excessive context length
   • Batch tool calls into a single AI invocation where possible
   • Reuse MCP connections; avoid频繁 creating and destroying connections

6. Parallel safety:

   • Cross-task data sharing MUST go through session_data; do not modify task_data and read it across tasks
   • Avoid multiple parallel tasks modifying the same session_data key simultaneously; use distributed locks when necessary

---

10. Exception Hierarchy

The framework provides fine-grained exception classification for implementing circuit-breaking, retries, and graceful degradation across different error scenarios:

TaskRouterError                 # Base framework exception (parent of all exceptions)
├── TaskRegistrationError       # Dynamic task/Prompt registration failure (e.g., validation error)
├── TaskExecutionError          # Task execution failure (business logic error or AI call exception)
├── TaskNotFoundError           # Specified task not found in registry
├── FunctionNotFoundError       # Specified function not found in registry
├── PromptNotFoundError         # Specified prompt not found in registry
├── DependencyCycleError        # Circular dependency detected in task graph (DAG validation failed)
├── RetryExhaustedError         # Retry attempts exhausted (timeout/network error retries failed)
├── ContextError                # Context operation exception (e.g., undefined variable, read-write conflict)
└── ProviderUnavailableError    # AI provider unavailable (e.g., invalid API key, service outage)

---

11. FAQ

Q1: What if a dynamically registered function has no custom handler?

A: The framework will use a default_handler (returns call parameter logs). For production, it is recommended to register a handler in advance or bind an existing function during dynamic registration.

Q2: How do I troubleshoot Redis connection failures?

A:

1. Verify Redis address, port, and password are correct
2. Check network connectivity and firewall rules
3. Enable DEBUG logging to inspect Redis connection error messages
4. Verify Redis service is running: redis-cli ping

Q3: What if a task gets stuck in a Tool Loop?

A:

1. Increase the max_tool_loops parameter (default is 3)
2. Check if the tool execution logic has an infinite loop
3. Verify that AI-returned tool_calls parameters conform to the function Schema
4. Enable DEBUG logging to inspect tool execution progress and return values

Q4: How do I make the AI remember long-term information across Tasks?

A: Write information that needs to be remembered long-term via Context.add_task_history_entry(ai_visible=True) to the global semantic memory. The AI will automatically read the most recent 3 entries in subsequent calls.

Q5: What should I watch out for in multi-process deployments?

A:

1. You MUST use RedisSessionStore; in-memory storage is not supported
2. MCP connections should use short-lived connections, or ensure connection isolation across processes
3. Dynamic registration operations require distributed locks to prevent duplicate registration
4. Avoid modifying the same global Registry configuration across multiple processes

---