Metadata-Version: 2.4
Name: triad-core-ai
Version: 0.1.0
Summary: A lightweight AI Agent orchestration framework with unified Task/Prompt/Function management and parallel routing
Author-email: Kenny Chen <geniusjees@hotmail.com>
License-Expression: MIT
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: jinja2>=3.1.6
Requires-Dist: mcp[cli]>=1.28.1
Requires-Dist: openai>=2.44.0
Requires-Dist: pydantic>=2.0.0
Requires-Dist: pytest>=9.1.1
Requires-Dist: pytest-asyncio>=1.4.0
Requires-Dist: redis>=8.0.1
Dynamic: license-file

# 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

```python
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

```python
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

```python
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.

```python
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.

```python
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.

```python
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:

```bash
pip install redis>=4.0
```

#### Basic Configuration (Standalone Redis)

```python
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)

```python
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

```python
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

```python
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

```python
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:

```python
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)

```python
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)

```python
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

```python
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

---

