agent-with-event-trigger¶
第4章 · 工具 · 配套项目
chapter4/agent-with-event-trigger
项目说明¶
Event-Triggered AI Agent with MCP Tools¶
A modern AI agent with native async support that responds to events from various sources. Built with FastAPI and integrated with 42 MCP tools for enhanced capabilities including browser automation, web search, document processing, and more.
🌟 Features¶
Core Capabilities¶
- ✅ Native Async - FastAPI with clean async/await support
- ✅ 42 MCP Tools - Automatically loaded from 3 MCP servers
- ✅ Event-Driven - Responds to web messages, emails, GitHub updates, timers
- ✅ System Hints - Timestamps, tool counters, TODO management
- ✅ Auto API Docs - Interactive Swagger UI at
/docs - ✅ Background Tasks - Process monitoring and system alerts
MCP Tool Categories¶
Collaboration Tools (18 tools): - Browser automation (navigate, screenshot, execute tasks) - Notifications (email, Telegram, Slack, Discord) - Human-in-the-loop (admin approval, input requests) - Timer management (one-time, recurring)
Execution Tools (6 tools): - File operations (write, edit with verification) - Code execution (Python interpreter, shell commands) - External integrations (Google Calendar, GitHub PRs)
Perception Tools (18 tools): - Web search and content extraction - Document reading (PDF, DOCX, PPTX) - Multimodal parsing (images, videos, webpages) - Public data (weather, stocks, Wikipedia, ArXiv) - Private data (Google Calendar, Notion)
⚡ 事件驱动演示（离线可运行，无需 API Key）¶
在启动 HTTP 服务器之前，推荐先运行 event_loop_demo.py，它在单个进程里
直观演示本章的核心概念——外部世界主动唤醒 Agent。脚本注册三类"事件触发器"，
由后台线程在事件真正发生时把结构化事件推入统一的事件队列，事件循环逐个取出
并唤醒 Agent 处理，形成"注册 → 触发 → 唤醒 → 处理"的完整闭环：
| 触发器 | 类 | 对应书中概念 |
|---|---|---|
| 一次性定时器 | OneShotTimer |
set_timer 一次性定时器（如"下周一 10:00 致电 DMV"） |
| 循环定时器 | RecurringTimer |
set_timer 循环定时器 / Heartbeat（如"每小时检查服务器"） |
| 文件监听 | FileWatchTrigger |
n8n 等平台的文件变更触发器 |
--mock 离线模式不调用大模型，用"模拟动作"打印 Agent 被唤醒后的处理过程，
因此无需任何 API Key 即可跑通：
# 离线演示全部触发器（一次性定时器 + 循环定时器 + 文件监听）
python event_loop_demo.py --mock
# 只演示一次性定时器；2 秒后触发，共运行 6 秒
python event_loop_demo.py --mock --trigger timer --delay 2 --duration 6
# 每 3 秒触发一次循环定时器
python event_loop_demo.py --mock --trigger recurring --interval 3 --duration 12
# 监听目录，向其中写入文件即可触发事件（另开终端 echo hello > watched_dir/a.txt）
python event_loop_demo.py --mock --trigger file --watch-dir watched_dir
离线运行的输出示例（节选）：
⏱️ [OneShotTimer(daily_backup_check)] 已注册：2 秒后触发
🔁 [RecurringTimer(health_check)] 已注册：每 3 秒触发一次
🟢 事件循环启动，将运行 8 秒，等待事件唤醒 Agent...
⚡ [OneShotTimer(daily_backup_check)] 触发事件 -> timer_trigger: 一次性定时器到期：请检查每日备份是否已经完成。
📥 事件循环取出第 1 个事件 -> 唤醒 Agent
🤖 Agent 被唤醒，收到消息: [Timer daily_backup_check triggered] 一次性定时器到期：请检查每日备份是否已经完成。
🛠️ [模拟动作] 读取定时任务上下文 -> 执行例行检查 -> 汇报结果
✅ Agent 处理完成: 已响应 timer_trigger 事件
去掉 --mock 即接入真实的大模型（默认仅用内置工具，不加载 MCP），
需要设置对应 provider 的 API Key：
OpenRouter 通用兜底：若所选 provider（默认
kimi）的 Key 缺失，但设置了OPENROUTER_API_KEY，event_loop_demo.py/server.py/quickstart.py会自动 改用openrouterprovider 继续运行（可用LLM_MODEL=openai/gpt-5.6-luna指定模型）。例如：OPENROUTER_API_KEY=sk-or-xxx LLM_MODEL=openai/gpt-5.6-luna python event_loop_demo.py --trigger timer
完整参数见 python event_loop_demo.py --help。
🚀 Quick Start¶
Installation¶
cd projects/week4/agent-with-event-trigger
# Install dependencies (includes FastAPI, uvicorn, MCP SDK)
pip install -r requirements.txt
# Set up environment
cp env.example .env
# Edit .env and add your API key
export KIMI_API_KEY='your-api-key-here'
Start the Server¶
服务器支持命令行参数（优先级高于环境变量），完整列表见 python server.py --help：
python server.py --port 9000 # 自定义端口
python server.py --provider doubao # 指定大模型提供商
python server.py --no-mcp # 只用内置工具，不加载 MCP 工具
Output:
🤖 EVENT-TRIGGERED AGENT SERVER (FastAPI)
✅ Starting server on port 8000
📡 API Documentation: http://localhost:8000/docs
📊 ReDoc: http://localhost:8000/redoc
🚀 Starting Event-Triggered Agent Server (FastAPI)
✅ Agent initialized with kimi provider
🔄 MCP tools enabled (default) - loading asynchronously...
✅ Discovered tools from 'collaboration': 18 tools
✅ Discovered tools from 'execution': 6 tools
✅ Discovered tools from 'perception': 18 tools
✅ MCP tools loaded: 42 tools available
✅ Server ready to receive events
INFO: Uvicorn running on http://0.0.0.0:8000
Interactive API Documentation¶
Visit http://localhost:8000/docs to: - 📖 Browse all available endpoints - 🧪 Test API calls interactively - 📝 See request/response schemas - ⚡ Send events with one click
📡 API Endpoints¶
Core Endpoints¶
# Health check
curl http://localhost:8000/health
# Check MCP tools status
curl http://localhost:8000/mcp/status
# Send an event
curl -X POST http://localhost:8000/event \
-H "Content-Type: application/json" \
-d '{
"event_type": "web_message",
"content": "Search the web for Python async best practices",
"metadata": {"user": "demo"}
}'
# Get agent status
curl http://localhost:8000/agent/status
# Reset agent state
curl -X POST http://localhost:8000/agent/reset
# Reload MCP tools
curl -X POST http://localhost:8000/mcp/reload
Using the Interactive Docs¶
- Open http://localhost:8000/docs
- Click on any endpoint (e.g.,
POST /event) - Click "Try it out"
- Fill in the request body
- Click "Execute"
- See the response instantly!
🎯 Usage Examples¶
Running the Standalone Example¶
For a complete demonstration of MCP integration without the server, run:
This standalone script: - Initializes the agent with MCP tools enabled - Loads all 42 tools from the 3 MCP servers - Processes a sample event (web search task) - Shows the complete flow from tool discovery to execution - Properly cleans up MCP connections
Output:
================================================================================
Event-Triggered Agent with MCP Tools Example
================================================================================
Initializing agent...
✅ Agent initialized with kimi provider
Loading MCP tools...
✅ Discovered tools from 'collaboration': 18 tools
✅ Discovered tools from 'execution': 6 tools
✅ Discovered tools from 'perception': 18 tools
✅ MCP tools loaded: 42 tools available
Testing Event Processing
================================================================================
📥 RECEIVED EVENT: Search the web for 'Python async best practices'...
This is useful for: - Testing MCP integration without running a server - Understanding the async tool loading flow - Debugging MCP connection issues - Learning how to use the agent programmatically
Example 1: Web Search Task¶
curl -X POST http://localhost:8000/event \
-H "Content-Type: application/json" \
-d '{
"event_type": "web_message",
"content": "Search for the latest FastAPI features and summarize them",
"metadata": {"user": "demo"}
}'
The agent will:
1. Use perception_web_search to find results
2. Parse the content with perception_webpage_reader
3. Summarize findings in the response
Example 2: Browser Automation¶
curl -X POST http://localhost:8000/event \
-H "Content-Type: application/json" \
-d '{
"event_type": "web_message",
"content": "Navigate to example.com and take a screenshot",
"metadata": {}
}'
Uses:
- collaboration_mcp_browser_navigate
- collaboration_mcp_browser_screenshot
Example 3: Document Processing¶
curl -X POST http://localhost:8000/event \
-H "Content-Type: application/json" \
-d '{
"event_type": "web_message",
"content": "Download and summarize the PDF from https://example.com/doc.pdf",
"metadata": {}
}'
Uses:
- perception_download
- perception_document_reader
Example 4: Email Notification¶
curl -X POST http://localhost:8000/event \
-H "Content-Type: application/json" \
-d '{
"event_type": "timer_trigger",
"content": "Send daily report to admin@example.com",
"metadata": {"scheduled": true}
}'
Uses:
- collaboration_mcp_send_email
⚙️ Configuration¶
Environment Variables¶
# Required
export KIMI_API_KEY="your-key"
# Optional
export LLM_PROVIDER="kimi" # kimi, siliconflow, doubao, openrouter
export LLM_MODEL="kimi-k3" # Override default model
export AGENT_PORT="8000" # Server port (default: 8000)
export ENABLE_MCP_TOOLS="true" # Enable MCP (default: true)
Disable MCP Tools¶
If you only want built-in tools:
Custom Port¶
🏗️ Architecture¶
┌─────────────────────────────────────────────────────────┐
│ FastAPI Server │
│ (Native Async) │
└───────────┬─────────────────────────────────────────────┘
│
├─► Lifespan Events (Startup/Shutdown)
│ └─► Load MCP Tools Asynchronously
│
├─► Event Handler (Process incoming events)
│ └─► EventTriggeredAgent
│ ├─► System Hints (timestamps, TODOs)
│ ├─► Tool Execution (MCP + built-in)
│ └─► Trajectory Saving
│
└─► MCP Server Manager
├─► Collaboration Tools (18 tools)
├─► Execution Tools (6 tools)
└─► Perception Tools (18 tools)
📂 Project Structure¶
agent-with-event-trigger/
├── agent.py # Event-triggered agent with system hints
├── event_types.py # Event type definitions
├── event_loop_demo.py # Offline event-loop demo (timer / recurring / file-watch triggers)
├── server.py # FastAPI server (main entry point)
├── requirements.txt # Dependencies (FastAPI, uvicorn, MCP)
├── env.example # Environment template
├── README.md # This file
├── FASTAPI_GUIDE.md # Detailed FastAPI guide
├── MCP_INTEGRATION.md # MCP tools documentation
└── example_with_mcp.py # Standalone MCP example
🔧 MCP Tools Reference¶
Check Available Tools¶
Response shows:
- tools: List of all 42 tool names
- tools_by_server: Tools grouped by server
- tools_count: Total count
- loaded: Whether MCP tools are active
Tool Naming Convention¶
MCP tools use underscore prefixes:
- collaboration_* - Collaboration tools
- execution_* - Execution tools
- perception_* - Perception tools
Built-in tools (no prefix):
- read_file
- write_file
- code_interpreter
- execute_command
- rewrite_todo_list
- update_todo_status
🚦 Event Types¶
class EventType(Enum):
# External input events
WEB_MESSAGE = "web_message" # Web interface
IM_MESSAGE = "im_message" # Instant messaging
EMAIL_REPLY = "email_reply" # Email responses
GITHUB_PR_UPDATE = "github_pr_update" # PR notifications
TIMER_TRIGGER = "timer_trigger" # Scheduled tasks (one-shot / recurring)
FILE_CHANGE = "file_change" # File watch trigger (created / modified)
# System reminder events
USER_TIMEOUT = "user_timeout" # No user activity
PROCESS_TIMEOUT = "process_timeout" # Long-running process
SYSTEM_ALERT = "system_alert" # System warnings
Event Format¶
{
"event_type": "web_message",
"content": "Your task description",
"metadata": {
"user_id": "user123",
"session_id": "session456"
}
}
🎨 Using the Client¶
The included client provides easy testing:
# Interactive mode
python client.py --mode interactive
# Test scenarios
python client.py --mode test
# Send a single event (defaults to web_message)
python client.py --message "Create a Python hello world script"
# Send a single event of a specific type
python client.py --event-type timer_trigger --message "检查每日备份"
🔐 Security Considerations¶
For production deployment:
- HTTPS: Use a reverse proxy (nginx, Caddy)
- Authentication: Add API key validation
- Rate Limiting: Prevent abuse
- Input Validation: Sanitize all inputs
- CORS: Configure allowed origins
- Environment: Use secrets management
🆚 Comparison: Flask vs FastAPI¶
| Feature | Old (Flask) | New (FastAPI) |
|---|---|---|
| Framework | Flask (WSGI) | FastAPI (ASGI) |
| Async Support | ❌ Threads | ✅ Native async/await |
| MCP Integration | ⚠️ Complex | ✅ Clean |
| API Docs | ❌ Manual | ✅ Auto-generated |
| Performance | Good | Better (2-3x) |
| Port | 4242 | 8000 |
| Deprecation Warnings | N/A | ✅ Fixed (lifespan) |
🐛 Troubleshooting¶
Port Already in Use¶
MCP Tools Not Loading¶
# Check status
curl http://localhost:8000/mcp/status
# Look for error in response
# Common issue: Missing API keys for MCP servers
# Reload tools
curl -X POST http://localhost:8000/mcp/reload
Import Errors¶
# Reinstall dependencies
pip install -r requirements.txt
# Verify FastAPI installed
python -c "import fastapi; print(fastapi.__version__)"
Agent Not Responding¶
# Check health
curl http://localhost:8000/health
# View logs in server terminal
# Check trajectory file: event_agent_trajectory.json
📚 Additional Documentation¶
- FASTAPI_GUIDE.md - Complete FastAPI setup guide
- MCP_INTEGRATION.md - MCP tools technical details
- QUICKSTART_MCP.md - Quick start with MCP
- README_MCP.md - MCP overview and troubleshooting
🔄 Migration from Old Version¶
If upgrading from Flask version:
- Port changed: 4242 → 8000
- No deprecation warnings: Using modern lifespan events
- MCP enabled by default: Set
ENABLE_MCP_TOOLS=falseto disable - Same API contracts: Endpoints work the same way
- Better performance: Native async for MCP calls
🤝 Contributing¶
Contributions welcome! Areas for improvement: - Additional MCP servers - More event types - Enhanced monitoring - Production-ready authentication - Kubernetes deployment configs
📄 License¶
MIT License - See LICENSE file for details
🙏 Acknowledgments¶
- Built with FastAPI
- MCP protocol by Model Context Protocol
- Tool servers: collaboration-tools, execution-tools, perception-tools
源代码¶
agent.py¶
"""
Event-Triggered AI Agent with System Hints
An agent that responds to events from various sources while maintaining
all the system hint features from the original implementation.
"""
import json
import os
import sys
import subprocess
import platform
import logging
import concurrent.futures
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime, timedelta
import requests
from openai import OpenAI
import traceback
import tempfile
import shutil
from pathlib import Path
from event_types import Event, EventType
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.types import TextContent
def _reasoning_safe_temperature(model, requested=1.0):
"""Reasoning models (Kimi K3, GPT-5, ...) only accept temperature=1.
Return 1 for those; otherwise the requested value so non-reasoning
providers (Doubao, DeepSeek, older Moonshot) are unchanged."""
m = str(model or "").lower().replace("/", "-")
return 1 if ("kimi-k3" in m or "gpt-5" in m) else requested
# Per-provider env var holding that provider's API key.
_PROVIDER_KEY_ENV = {
"siliconflow": "SILICONFLOW_API_KEY",
"doubao": "DOUBAO_API_KEY",
"kimi": "KIMI_API_KEY",
"moonshot": "KIMI_API_KEY",
"openrouter": "OPENROUTER_API_KEY",
}
def resolve_provider_and_key(provider: Optional[str] = None):
"""Resolve (provider, api_key) applying a universal OpenRouter fallback.
Preferred provider (default from LLM_PROVIDER, else 'kimi') is used as today
when its own key is present. Otherwise, if OPENROUTER_API_KEY is set, fall
back to the already-supported 'openrouter' provider so the agent still runs.
Returns (provider, None) when no usable key is found, leaving the caller to
emit its own error.
"""
provider = (provider or os.getenv("LLM_PROVIDER", "kimi")).lower()
key_env = _PROVIDER_KEY_ENV.get(provider)
api_key = os.getenv(key_env) if key_env else None
if api_key:
return provider, api_key
or_key = os.getenv("OPENROUTER_API_KEY")
if or_key:
return "openrouter", or_key
return provider, None
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class TodoStatus(Enum):
"""Status of a TODO item"""
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
CANCELLED = "cancelled"
@dataclass
class TodoItem:
"""Represents a single TODO item"""
id: int
content: str
status: TodoStatus = TodoStatus.PENDING
created_at: str = field(default_factory=lambda: datetime.now().isoformat())
updated_at: Optional[str] = None
@dataclass
class ToolCall:
"""Represents a single tool call with enhanced tracking"""
tool_name: str
arguments: Dict[str, Any]
result: Optional[Any] = None
error: Optional[str] = None
call_number: int = 1
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
duration_ms: Optional[int] = None
@dataclass
class SystemHintConfig:
"""Configuration for system hints"""
enable_timestamps: bool = True
enable_tool_counter: bool = True
enable_todo_list: bool = True
enable_detailed_errors: bool = True
enable_system_state: bool = True
timestamp_format: str = "%Y-%m-%d %H:%M:%S"
simulate_time_delay: bool = False
save_trajectory: bool = True
trajectory_file: str = "trajectory.json"
# Model configuration (matching conversational_agent.py)
temperature: float = 0.7
max_tokens: int = 4096
# MCP server configuration
use_mcp_servers: bool = False # Disabled by default - requires async setup
mcp_collaboration_tools_path: str = "../collaboration-tools/src/main.py"
mcp_execution_tools_path: str = "../execution-tools/server.py"
mcp_perception_tools_path: str = "../perception-tools/src/main.py"
class MCPServerManager:
"""Manages connections to multiple MCP servers"""
def __init__(self):
self.sessions: Dict[str, ClientSession] = {}
self.tools: Dict[str, Any] = {}
self.server_contexts = [] # Store context managers for proper cleanup
async def connect_server(self, name: str, script_path: str) -> bool:
"""
Connect to an MCP server with proper error isolation
NOTE: This stores tool metadata only. For actual tool execution,
you'll need to spawn MCP servers differently or use built-in tools.
Args:
name: Server name
script_path: Path to the MCP server script
Returns:
True if connection successful, False otherwise
"""
try:
# Check if script exists
if not os.path.exists(script_path):
logger.warning(f"MCP server script not found: {script_path}")
return False
logger.info(f"Discovering tools from MCP server '{name}' at {script_path}")
server_params = StdioServerParameters(
command=sys.executable,
args=[script_path],
env=os.environ.copy() # Pass through environment variables
)
# Use a temporary connection just to discover tools
# The actual tool execution will spawn servers on-demand
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# Discover tools
tools_result = await session.list_tools()
tools = tools_result.tools
# Store tool metadata (not live session)
for tool in tools:
# Use underscore separator for valid function names
# OpenAI-compatible APIs reject names with dots
tool_key = f"{name}_{tool.name}"
self.tools[tool_key] = {
"server": name,
"tool": tool,
"script_path": script_path,
"server_params": server_params
}
logger.info(f"✅ Discovered tools from '{name}': {len(tools)} tools")
return True
except Exception as e:
logger.warning(f"Failed to discover tools from '{name}': {str(e)[:100]}")
return False
async def call_tool(self, tool_key: str, arguments: Dict[str, Any]) -> Any:
"""
Call an MCP tool by spawning a fresh server connection
Args:
tool_key: Tool key in format "server.tool_name"
arguments: Tool arguments
Returns:
Tool result
"""
if tool_key not in self.tools:
raise ValueError(f"Unknown MCP tool: {tool_key}")
tool_info = self.tools[tool_key]
tool = tool_info["tool"]
server_params = tool_info["server_params"]
try:
# Spawn fresh connection for this tool call
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
result = await session.call_tool(tool.name, arguments)
# Extract text content from result
text_content = []
if hasattr(result, 'content'):
for c in result.content:
if isinstance(c, TextContent):
text_content.append(c.text)
return {
"success": True,
"result": "\n".join(text_content) if text_content else str(result)
}
except Exception as e:
return {
"success": False,
"error": str(e)
}
async def disconnect_all(self):
"""
Cleanup MCP manager (no persistent connections to close)
Since we spawn fresh connections for each tool call,
there's nothing to disconnect.
"""
logger.info("MCP manager cleanup complete (no persistent connections)")
self.sessions.clear()
self.tools.clear()
self.server_contexts.clear()
class EventTriggeredAgent:
"""
Event-Triggered AI Agent with System Hints
Responds to events while maintaining all system hint capabilities
"""
def __init__(self, api_key: str, provider: str = "kimi",
model: Optional[str] = None, config: Optional[SystemHintConfig] = None,
verbose: bool = True):
"""
Initialize the event-triggered agent
Args:
api_key: API key for the LLM provider
provider: LLM provider ('siliconflow', 'doubao', 'kimi', 'moonshot', 'openrouter')
model: Optional model override
config: System hint configuration
verbose: If True, log full details
"""
self.provider = provider.lower()
self.verbose = verbose
self.config = config or SystemHintConfig()
# Configure client based on provider (matching conversational_agent.py)
if self.provider == "siliconflow":
self.client = OpenAI(
api_key=api_key,
base_url="https://api.siliconflow.cn/v1"
)
self.model = model or "Qwen/Qwen3-235B-A22B-Thinking-2507"
elif self.provider == "doubao":
self.client = OpenAI(
api_key=api_key,
base_url="https://ark.cn-beijing.volces.com/api/v3"
)
self.model = model or "doubao-seed-1-6-thinking-250715"
elif self.provider == "kimi" or self.provider == "moonshot":
self.client = OpenAI(
api_key=api_key,
base_url="https://api.moonshot.cn/v1"
)
self.model = model or "kimi-k3"
elif self.provider == "openrouter":
self.client = OpenAI(
api_key=api_key,
base_url="https://openrouter.ai/api/v1"
)
# Default to Gemini 3.5 Flash, but allow any of the supported models
self.model = model or "google/gemini-3.5-flash"
# Supported models: google/gemini-3.5-flash, openai/gpt-5.6-luna, anthropic/claude-sonnet-4.6
else:
raise ValueError(f"Unsupported provider: {provider}. Use 'siliconflow', 'doubao', 'kimi', 'moonshot', or 'openrouter'")
# Initialize tracking
self.tool_call_counts: Dict[str, int] = {}
self.tool_calls: List[ToolCall] = []
self.todo_list: List[TodoItem] = []
self.next_todo_id = 1
# Initialize conversation history
self.conversation_history = []
self.simulated_time = datetime.now()
self._init_system_prompt()
# Track current working directory
self.current_directory = os.getcwd()
# Event tracking
self.last_user_interaction = datetime.now()
self.background_processes: Dict[str, Dict[str, Any]] = {}
# Initialize MCP server manager
self.mcp_manager = MCPServerManager()
self.mcp_tools_loaded = False
logger.info(f"Event-Triggered Agent initialized with provider: {self.provider}, model: {self.model}")
logger.info("Note: Call load_mcp_tools() to connect to MCP servers")
async def load_mcp_tools(self):
"""
Load tools from MCP servers
This must be called in an async context after agent initialization.
"""
if not self.config.use_mcp_servers:
logger.info("MCP servers disabled in config")
return
logger.info("Loading tools from MCP servers...")
# Get the directory where the agent script is located
agent_dir = os.path.dirname(os.path.abspath(__file__))
# Try to connect to collaboration-tools
collab_path = os.path.join(agent_dir, self.config.mcp_collaboration_tools_path)
collab_loaded = await self.mcp_manager.connect_server("collaboration", collab_path)
# Try to connect to execution-tools
exec_path = os.path.join(agent_dir, self.config.mcp_execution_tools_path)
exec_loaded = await self.mcp_manager.connect_server("execution", exec_path)
# Try to connect to perception-tools
percept_path = os.path.join(agent_dir, self.config.mcp_perception_tools_path)
percept_loaded = await self.mcp_manager.connect_server("perception", percept_path)
# Set flag if any tools were loaded
self.mcp_tools_loaded = collab_loaded or exec_loaded or percept_loaded
if self.mcp_tools_loaded:
logger.info(f"✅ MCP tools loaded: {len(self.mcp_manager.tools)} tools available")
logger.info(f" Available MCP tools: {list(self.mcp_manager.tools.keys())[:5]}...")
else:
logger.info("⚠️ No MCP servers found, using built-in tools only")
def _init_system_prompt(self):
"""Initialize the system prompt for the conversation"""
system_content = """You are an intelligent assistant with access to various tools for file operations, code execution, and system commands.
You respond to events from multiple sources including:
- User messages from web interfaces and instant messaging
- Email replies and GitHub notifications
- System reminders and timeout alerts
- Timer triggers and process monitoring events
Your task is to complete the given objectives efficiently using the available tools. Think step by step and use tools as needed.
## TODO List Management Rules:
- For any complex task with 3+ distinct steps, immediately create a TODO list using `rewrite_todo_list`
- Break down the user's request into specific, actionable TODO items
- Update TODO items to 'in_progress' when starting work on them using `update_todo_status`
- Mark items as 'completed' immediately after finishing them
- Only have ONE item 'in_progress' at a time
- If you encounter errors or need to change approach, update relevant TODOs to 'cancelled' and add new ones
- Use the TODO list as your primary planning and tracking mechanism
- Reference TODO items by their ID when discussing progress
## Key Behaviors:
1. ALWAYS start complex tasks by creating a TODO list
2. Pay attention to timestamps to understand the timeline of events
3. Notice tool call numbers (e.g., "Tool call #3") to avoid repetitive loops - if you see high numbers, change strategy
4. Learn from detailed error messages to fix issues and adapt your approach
5. Be aware of your current directory and system environment shown in system state
6. When exploring projects, systematically read key files (README, main.py, agent.py) to understand structure
## Event Response Guidelines:
- Acknowledge the event source in your response
- For timeout events, proactively check status and take appropriate action
- For system alerts, investigate the issue before responding
- Maintain context across multiple events from the same conversation
## Error Handling:
- Read error messages carefully - they contain specific information about what went wrong
- Use the suggestions provided in error messages to fix issues
- If a tool fails multiple times (check the call number), try a different approach
- Common fixes: check file paths, verify current directory, ensure proper permissions
Important: When you have completed all tasks, clearly state "FINAL ANSWER:" followed by a comprehensive summary of what was accomplished."""
self.conversation_history = [
{
"role": "system",
"content": system_content
}
]
def _get_system_state(self) -> str:
"""Get current system state information"""
if not self.config.enable_system_state:
return ""
# Detect OS
system = platform.system()
if system == "Windows":
shell_type = "Windows Command Prompt or PowerShell"
elif system == "Darwin":
shell_type = "macOS Terminal (zsh/bash)"
else:
shell_type = f"Linux Shell ({os.environ.get('SHELL', 'bash')})"
state_info = [
f"Current Time: {self._get_timestamp()}",
f"Current Directory: {self.current_directory}",
f"System: {system} ({platform.release()})",
f"Shell Environment: {shell_type}",
f"Python Version: {sys.version.split()[0]}"
]
# Add background process info if any
if self.background_processes:
state_info.append(f"Background Processes: {len(self.background_processes)} active")
return "\n".join(state_info)
def _get_timestamp(self) -> str:
"""Get formatted timestamp"""
if self.config.simulate_time_delay:
return self.simulated_time.strftime(self.config.timestamp_format)
return datetime.now().strftime(self.config.timestamp_format)
def _advance_simulated_time(self, hours: int = 0, minutes: int = 0, seconds: int = 30):
"""Advance simulated time for demo purposes"""
if self.config.simulate_time_delay:
self.simulated_time += timedelta(hours=hours, minutes=minutes, seconds=seconds)
def _save_trajectory(self, iteration: int, final_answer: Optional[str] = None):
"""Save current trajectory to JSON file for debugging"""
if not self.config.save_trajectory:
return
trajectory_data = {
"timestamp": datetime.now().isoformat(),
"iteration": iteration,
"provider": self.provider,
"model": self.model,
"conversation_history": self.conversation_history,
"tool_calls": [
{
"tool_name": call.tool_name,
"arguments": call.arguments,
"result": call.result,
"error": call.error,
"call_number": call.call_number,
"timestamp": call.timestamp,
"duration_ms": call.duration_ms
}
for call in self.tool_calls
],
"todo_list": [
{
"id": item.id,
"content": item.content,
"status": item.status.value,
"created_at": item.created_at,
"updated_at": item.updated_at
}
for item in self.todo_list
],
"current_directory": self.current_directory,
"final_answer": final_answer,
"background_processes": self.background_processes,
"config": {
"enable_timestamps": self.config.enable_timestamps,
"enable_tool_counter": self.config.enable_tool_counter,
"enable_todo_list": self.config.enable_todo_list,
"enable_detailed_errors": self.config.enable_detailed_errors,
"enable_system_state": self.config.enable_system_state,
"timestamp_format": self.config.timestamp_format,
"simulate_time_delay": self.config.simulate_time_delay
}
}
try:
with open(self.config.trajectory_file, 'w', encoding='utf-8') as f:
json.dump(trajectory_data, f, indent=2, ensure_ascii=False)
if self.verbose:
logger.info(f"Trajectory saved to {self.config.trajectory_file} (iteration {iteration})")
except Exception as e:
logger.warning(f"Failed to save trajectory: {e}")
def _format_todo_list(self) -> str:
"""Format TODO list for display"""
if not self.todo_list:
return "TODO List: Empty"
lines = ["TODO List:"]
for item in self.todo_list:
status_symbol = {
TodoStatus.PENDING: "⏳",
TodoStatus.IN_PROGRESS: "🔄",
TodoStatus.COMPLETED: "✅",
TodoStatus.CANCELLED: "❌"
}.get(item.status, "❓")
lines.append(f" [{item.id}] {status_symbol} {item.content} ({item.status.value})")
return "\n".join(lines)
def _get_system_hint(self) -> Optional[str]:
"""Get system hint content with current state"""
if not any([self.config.enable_system_state, self.config.enable_todo_list]):
return None
hint_parts = []
if self.config.enable_system_state:
hint_parts.append("=== SYSTEM STATE ===")
hint_parts.append(self._get_system_state())
hint_parts.append("")
if self.config.enable_todo_list and self.todo_list:
hint_parts.append("=== CURRENT TASKS ===")
hint_parts.append(self._format_todo_list())
hint_parts.append("")
if hint_parts:
return "\n".join(hint_parts)
return None
def _get_tools_description(self) -> List[Dict[str, Any]]:
"""Get tool descriptions for the model"""
tools = []
# Add MCP tools if available
if self.mcp_tools_loaded:
for tool_key, tool_info in self.mcp_manager.tools.items():
mcp_tool = tool_info["tool"]
# Convert MCP tool schema to OpenAI function format
tool_desc = {
"type": "function",
"function": {
"name": tool_key, # Use prefixed name (e.g., "collaboration.mcp_browser_navigate")
"description": mcp_tool.description or mcp_tool.name,
"parameters": mcp_tool.inputSchema if hasattr(mcp_tool, 'inputSchema') else {
"type": "object",
"properties": {}
}
}
}
tools.append(tool_desc)
else:
# Fallback to built-in tools if MCP servers not available
tools.extend([
{
"type": "function",
"function": {
"name": "read_file",
"description": "Read the contents of a text file. Returns error for binary files. Supports partial reading for large files.",
"parameters": {
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "Path to the file to read (absolute or relative to current directory)"
},
"begin_line": {
"type": "integer",
"description": "Optional: Line number to start reading from (1-based indexing)"
},
"number_lines": {
"type": "integer",
"description": "Optional: Number of lines to read from begin_line"
}
},
"required": ["file_path"]
}
}
},
{
"type": "function",
"function": {
"name": "write_file",
"description": "Write content to a file (creates or overwrites)",
"parameters": {
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "Path to the file to write"
},
"content": {
"type": "string",
"description": "Content to write to the file"
}
},
"required": ["file_path", "content"]
}
}
},
{
"type": "function",
"function": {
"name": "code_interpreter",
"description": "Execute Python code in a restricted environment",
"parameters": {
"type": "object",
"properties": {
"code": {
"type": "string",
"description": "Python code to execute"
}
},
"required": ["code"]
}
}
},
{
"type": "function",
"function": {
"name": "execute_command",
"description": "Execute a shell command in the current directory",
"parameters": {
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "Shell command to execute"
},
"working_dir": {
"type": "string",
"description": "Optional working directory for the command"
}
},
"required": ["command"]
}
}
}
])
# Always add TODO management tools if enabled
if self.config.enable_todo_list:
tools.extend([
{
"type": "function",
"function": {
"name": "rewrite_todo_list",
"description": "Rewrite the TODO list with new pending items (keeps completed/cancelled items)",
"parameters": {
"type": "object",
"properties": {
"items": {
"type": "array",
"items": {
"type": "string"
},
"description": "List of new TODO items to add as pending"
}
},
"required": ["items"]
}
}
},
{
"type": "function",
"function": {
"name": "update_todo_status",
"description": "Update the status of existing TODO items",
"parameters": {
"type": "object",
"properties": {
"updates": {
"type": "array",
"items": {
"type": "object",
"properties": {
"id": {
"type": "integer",
"description": "TODO item ID"
},
"status": {
"type": "string",
"enum": ["pending", "in_progress", "completed", "cancelled"],
"description": "New status for the item"
}
},
"required": ["id", "status"]
},
"description": "List of TODO items to update with their new status"
}
},
"required": ["updates"]
}
}
}
])
return tools
def _execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Tuple[Any, Optional[str]]:
"""Execute a tool and return the result with detailed error information"""
start_time = datetime.now()
try:
# Check if it's an MCP tool (prefixed with server name using underscore)
if "_" in tool_name and tool_name in self.mcp_manager.tools:
# Execute MCP tool asynchronously
# Check if there's already a running event loop
try:
loop = asyncio.get_running_loop()
# If we're already in an async context, we can't use asyncio.run()
# Create a new event loop in a separate thread
with concurrent.futures.ThreadPoolExecutor() as executor:
future = executor.submit(
asyncio.run,
self.mcp_manager.call_tool(tool_name, arguments)
)
result = future.result()
except RuntimeError:
# No event loop running, safe to use asyncio.run()
result = asyncio.run(self.mcp_manager.call_tool(tool_name, arguments))
duration_ms = int((datetime.now() - start_time).total_seconds() * 1000)
return result, None if result.get("success") else result.get("error")
# Built-in tools
if tool_name == "read_file":
result = self._tool_read_file(**arguments)
elif tool_name == "write_file":
result = self._tool_write_file(**arguments)
elif tool_name == "code_interpreter":
result = self._tool_code_interpreter(**arguments)
elif tool_name == "execute_command":
result = self._tool_execute_command(**arguments)
elif tool_name == "rewrite_todo_list":
result = self._tool_rewrite_todo_list(**arguments)
elif tool_name == "update_todo_status":
result = self._tool_update_todo_status(**arguments)
else:
error = f"Unknown tool: {tool_name}"
return {"error": error}, error
duration_ms = int((datetime.now() - start_time).total_seconds() * 1000)
return result, None
except Exception as e:
duration_ms = int((datetime.now() - start_time).total_seconds() * 1000)
error_detail = self._get_detailed_error(e, tool_name, arguments)
if self.config.enable_detailed_errors:
return {"error": error_detail}, error_detail
else:
return {"error": str(e)}, str(e)
def _get_detailed_error(self, exception: Exception, tool_name: str, arguments: Dict[str, Any]) -> str:
"""Get detailed error information for debugging"""
error_parts = [
f"Tool '{tool_name}' failed with {type(exception).__name__}: {str(exception)}",
f"Arguments: {json.dumps(arguments, indent=2)}",
]
if self.verbose:
tb = traceback.format_exc()
error_parts.append(f"Traceback:\n{tb}")
suggestions = self._get_error_suggestions(exception, tool_name)
if suggestions:
error_parts.append(f"Suggestions: {suggestions}")
return "\n".join(error_parts)
def _get_error_suggestions(self, exception: Exception, tool_name: str) -> str:
"""Get suggestions for fixing common errors"""
error_str = str(exception).lower()
exception_type = type(exception).__name__
suggestions = []
if "permission" in error_str or exception_type == "PermissionError":
suggestions.append("Check file/directory permissions")
suggestions.append("Try using a different directory or running with appropriate permissions")
elif "not found" in error_str or "no such file" in error_str or exception_type == "FileNotFoundError":
suggestions.append("Verify the file/directory path exists")
suggestions.append("Check the current working directory")
suggestions.append("Use absolute paths or create the file/directory first")
elif "syntax" in error_str or exception_type == "SyntaxError":
suggestions.append("Check the code syntax")
suggestions.append("Ensure proper indentation and valid Python syntax")
elif "timeout" in error_str:
suggestions.append("The operation took too long")
suggestions.append("Try with simpler input or break into smaller steps")
elif "import" in error_str or exception_type == "ImportError":
suggestions.append("Required module not available in restricted environment")
suggestions.append("Use only built-in Python modules")
return " | ".join(suggestions) if suggestions else ""
# Tool implementations (copied from original agent.py)
def _tool_read_file(self, file_path: str, begin_line: Optional[int] = None,
number_lines: Optional[int] = None) -> Dict[str, Any]:
"""Read file contents with optional line-based reading"""
try:
if not os.path.isabs(file_path):
file_path = os.path.join(self.current_directory, file_path)
if not os.path.exists(file_path):
raise FileNotFoundError(f"File not found: {file_path}")
# Check if it's a binary file
try:
with open(file_path, 'rb') as f:
chunk = f.read(1024)
if b'\x00' in chunk:
return {
"success": False,
"error": "Cannot read binary file. This tool only supports text files.",
"file_path": file_path,
"is_binary": True
}
try:
chunk.decode('utf-8')
except UnicodeDecodeError:
return {
"success": False,
"error": "File is not a valid text file (encoding error).",
"file_path": file_path,
"is_binary": True
}
except Exception as e:
raise
with open(file_path, 'r', encoding='utf-8') as f:
if begin_line is not None or number_lines is not None:
all_lines = f.readlines()
total_lines = len(all_lines)
start_line = (begin_line - 1) if begin_line is not None else 0
if start_line < 0:
start_line = 0
if start_line >= total_lines:
return {
"success": False,
"error": f"begin_line {begin_line} is beyond file length ({total_lines} lines)",
"file_path": file_path,
"total_lines": total_lines
}
if number_lines is not None:
end_line = min(start_line + number_lines, total_lines)
else:
end_line = total_lines
selected_lines = all_lines[start_line:end_line]
content = ''.join(selected_lines)
stat = os.stat(file_path)
return {
"success": True,
"file_path": file_path,
"content": content,
"size_bytes": stat.st_size,
"total_lines": total_lines,
"begin_line": start_line + 1,
"end_line": end_line,
"lines_read": len(selected_lines),
"partial_read": True
}
else:
content = f.read()
stat = os.stat(file_path)
return {
"success": True,
"file_path": file_path,
"content": content,
"size_bytes": stat.st_size,
"lines": len(content.splitlines()),
"partial_read": False
}
except Exception as e:
raise
def _tool_write_file(self, file_path: str, content: str) -> Dict[str, Any]:
"""Write content to file"""
try:
if not os.path.isabs(file_path):
file_path = os.path.join(self.current_directory, file_path)
os.makedirs(os.path.dirname(file_path), exist_ok=True)
with open(file_path, 'w', encoding='utf-8') as f:
f.write(content)
return {
"success": True,
"file_path": file_path,
"bytes_written": len(content.encode('utf-8')),
"lines_written": len(content.splitlines())
}
except Exception as e:
raise
def _tool_code_interpreter(self, code: str) -> Dict[str, Any]:
"""Execute Python code in restricted environment"""
try:
import io
import contextlib
output_buffer = io.StringIO()
error_buffer = io.StringIO()
with contextlib.redirect_stdout(output_buffer), contextlib.redirect_stderr(error_buffer):
exec(code)
stdout = output_buffer.getvalue()
stderr = error_buffer.getvalue()
return {
"success": True,
"stdout": stdout,
"stderr": stderr,
}
except Exception as e:
raise
def _tool_execute_command(self, command: str, working_dir: Optional[str] = None) -> Dict[str, Any]:
"""Execute shell command"""
try:
if working_dir is None:
working_dir = self.current_directory
elif not os.path.isabs(working_dir):
working_dir = os.path.join(self.current_directory, working_dir)
if command.strip().startswith('cd '):
new_dir = command.strip()[3:].strip()
if not os.path.isabs(new_dir):
new_dir = os.path.join(self.current_directory, new_dir)
if os.path.isdir(new_dir):
self.current_directory = os.path.abspath(new_dir)
return {
"success": True,
"command": command,
"output": f"Changed directory to: {self.current_directory}",
"return_code": 0
}
else:
raise FileNotFoundError(f"Directory not found: {new_dir}")
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
cwd=working_dir,
timeout=30
)
return {
"success": result.returncode == 0,
"command": command,
"output": result.stdout,
"error": result.stderr if result.stderr else None,
"return_code": result.returncode,
"working_dir": working_dir
}
except subprocess.TimeoutExpired:
raise TimeoutError(f"Command timed out after 30 seconds: {command}")
except Exception as e:
raise
def _tool_rewrite_todo_list(self, items: List[str]) -> Dict[str, Any]:
"""Rewrite TODO list with new pending items"""
kept_items = [
item for item in self.todo_list
if item.status in [TodoStatus.COMPLETED, TodoStatus.CANCELLED]
]
new_items = []
for content in items:
new_items.append(TodoItem(
id=self.next_todo_id,
content=content,
status=TodoStatus.PENDING
))
self.next_todo_id += 1
self.todo_list = kept_items + new_items
return {
"success": True,
"kept_items": len(kept_items),
"new_items": len(new_items),
"total_items": len(self.todo_list)
}
def _tool_update_todo_status(self, updates: List[Dict[str, Any]]) -> Dict[str, Any]:
"""Update status of TODO items"""
updated_count = 0
for update in updates:
item_id = update["id"]
new_status = TodoStatus(update["status"])
for item in self.todo_list:
if item.id == item_id:
item.status = new_status
item.updated_at = datetime.now().isoformat()
updated_count += 1
break
return {
"success": True,
"updated_items": updated_count,
"total_items": len(self.todo_list)
}
def handle_event(self, event: Event, max_iterations: int = 20) -> Dict[str, Any]:
"""
Handle an incoming event and generate a response
Args:
event: The event to handle
max_iterations: Maximum number of tool call iterations
Returns:
Response with agent's actions and final answer
"""
logger.info(f"\n{'='*80}")
logger.info(f"📥 RECEIVED EVENT")
logger.info(f"{'='*80}")
logger.info(f"Event Type: {event.event_type.value}")
logger.info(f"Timestamp: {event.timestamp}")
logger.info(f"Content: {event.content}")
if event.metadata:
logger.info(f"Metadata: {json.dumps(event.metadata, indent=2)}")
logger.info(f"{'='*80}\n")
# Convert event to user message
user_message = event.to_user_message()
# Add timestamp prefix if enabled
if self.config.enable_timestamps:
timestamp_prefix = f"[{self._get_timestamp()}] "
user_message = timestamp_prefix + user_message
# Update last user interaction time for external events
if event.event_type in [EventType.WEB_MESSAGE, EventType.IM_MESSAGE, EventType.EMAIL_REPLY]:
self.last_user_interaction = datetime.now()
# Add user message to conversation
self.conversation_history.append({"role": "user", "content": user_message})
iteration = 0
final_answer = None
while iteration < max_iterations:
iteration += 1
logger.info(f"Iteration {iteration}/{max_iterations}")
self._advance_simulated_time(seconds=5)
self._save_trajectory(iteration)
try:
messages_to_send = self.conversation_history.copy()
system_hint = self._get_system_hint()
if system_hint:
messages_to_send.append({"role": "user", "content": system_hint})
response = self.client.chat.completions.create(
model=self.model,
messages=messages_to_send,
tools=self._get_tools_description(),
tool_choice="auto",
temperature=_reasoning_safe_temperature(self.model, self.config.temperature),
max_tokens=self.config.max_tokens
)
message = response.choices[0].message
has_tool_calls = bool(getattr(message, "tool_calls", None))
# Terminal path: a text reply with no tool calls ends the loop,
# even without the FINAL ANSWER: marker (e.g. a plain "hi"
# reply). Previously only "FINAL ANSWER:" broke the loop, so
# plain replies were re-sent for up to max_iterations.
if not has_tool_calls:
self.conversation_history.append(message.model_dump())
content = (message.content or "").strip()
if content:
final_answer = (content.split("FINAL ANSWER:", 1)[1].strip()
if "FINAL ANSWER:" in content else content)
logger.info(f"✅ Terminal text response (no tool calls); final answer: {final_answer[:100]}...")
else:
logger.warning("Empty model response with no tool calls; "
"stopping to avoid burning remaining iterations")
self._save_trajectory(iteration, final_answer)
break
if has_tool_calls:
self.conversation_history.append(message.model_dump())
for tool_call in message.tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
if self.config.enable_tool_counter:
self.tool_call_counts[function_name] = self.tool_call_counts.get(function_name, 0) + 1
call_number = self.tool_call_counts[function_name]
else:
call_number = 1
logger.info(f"🔧 Executing tool: {function_name} (call #{call_number})")
args_str = json.dumps(function_args)
if len(args_str) > 200:
logger.info(f" 📥 Args: {args_str[:200]}...")
else:
logger.info(f" 📥 Args: {args_str}")
result, error = self._execute_tool(function_name, function_args)
if error:
error_preview = str(error).replace('\n', ' ')[:150]
logger.info(f" ❌ Error: {error_preview}")
else:
if isinstance(result, dict):
if result.get('success'):
if 'output' in result and result['output']:
output_preview = str(result['output']).replace('\n', ' ')[:100]
logger.info(f" ✅ Success: {output_preview}...")
elif 'content' in result:
if result.get('partial_read'):
logger.info(f" ✅ Success: Read lines {result.get('begin_line', 1)}-{result.get('end_line', 0)} "
f"({result.get('lines_read', 0)} lines) from {result.get('total_lines', 0)} total")
else:
logger.info(f" ✅ Success: Read {result.get('lines', 0)} lines, {result.get('size_bytes', 0)} bytes")
elif 'file_path' in result:
logger.info(f" ✅ Success: File operation on {result['file_path']}")
else:
logger.info(f" ✅ Success: Operation completed")
elif result.get('success') is False:
if result.get('is_binary'):
logger.info(f" ⚠️ Binary file detected: {result.get('file_path', 'unknown')}")
else:
logger.info(f" ⚠️ Failed: {result.get('error', 'Unknown error')[:100]}")
else:
logger.info(f" ✅ Success: Operation completed")
else:
result_preview = str(result).replace('\n', ' ')[:150]
logger.info(f" ✅ Result: {result_preview}")
tool_call_record = ToolCall(
tool_name=function_name,
arguments=function_args,
result=result if not error else None,
error=error,
call_number=call_number
)
self.tool_calls.append(tool_call_record)
tool_content = json.dumps(result)
metadata_parts = []
if self.config.enable_timestamps:
metadata_parts.append(f"[{self._get_timestamp()}]")
if self.config.enable_tool_counter:
metadata_parts.append(f"[Tool call #{call_number} for '{function_name}']")
if metadata_parts:
tool_content = " ".join(metadata_parts) + "\n" + tool_content
self.conversation_history.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": tool_content
})
# If the same turn also tagged FINAL ANSWER: (unusual with
# tool calls), still stop after recording the tool results.
if message.content and "FINAL ANSWER:" in message.content:
final_answer = message.content.split("FINAL ANSWER:", 1)[1].strip()
logger.info(f"✅ Final answer found alongside tool calls: {final_answer[:100]}...")
self._save_trajectory(iteration, final_answer)
break
except Exception as e:
logger.error(f"Error during event handling: {str(e)}")
self._save_trajectory(iteration)
return {
"success": False,
"error": str(e),
"tool_calls": self.tool_calls,
"iterations": iteration,
"trajectory_file": self.config.trajectory_file if self.config.save_trajectory else None
}
self._save_trajectory(iteration, final_answer)
logger.info(f"\n{'='*80}")
logger.info(f"📤 AGENT RESPONSE")
logger.info(f"{'='*80}")
if final_answer:
logger.info(f"Response: {final_answer}")
else:
logger.info(f"Response: Task processing completed ({iteration} iterations)")
logger.info(f"Tool Calls: {len(self.tool_calls)}")
logger.info(f"{'='*80}\n")
return {
"final_answer": final_answer,
"tool_calls": self.tool_calls,
"todo_list": [
{
"id": item.id,
"content": item.content,
"status": item.status.value
}
for item in self.todo_list
],
"iterations": iteration,
"success": final_answer is not None,
"trajectory_file": self.config.trajectory_file if self.config.save_trajectory else None
}
def reset(self):
"""Reset the agent's state"""
self.tool_call_counts = {}
self.tool_calls = []
self.todo_list = []
self.next_todo_id = 1
self.current_directory = os.getcwd()
self.simulated_time = datetime.now()
self.last_user_interaction = datetime.now()
self.background_processes = {}
self._init_system_prompt()
logger.info("Agent state reset")
def __del__(self):
"""Cleanup when agent is destroyed"""
if hasattr(self, 'mcp_manager') and self.mcp_manager.sessions:
try:
asyncio.run(self.mcp_manager.disconnect_all())
except Exception as e:
logger.warning(f"Error disconnecting MCP servers: {e}")
client.py¶
"""
Event Client - Send test events to the event-triggered agent
"""
import requests
import json
import time
import argparse
from datetime import datetime
from event_types import EventType
class EventClient:
"""Client to send events to the event-triggered agent server"""
def __init__(self, server_url: str = "http://localhost:8000"):
"""
Initialize the client
Args:
server_url: URL of the event server
"""
self.server_url = server_url.rstrip('/')
def send_event(self, event_type: str, content: str, metadata: dict = None) -> dict:
"""
Send an event to the agent
Args:
event_type: Type of event (e.g., 'web_message', 'im_message')
content: Content of the event
metadata: Additional metadata for the event
Returns:
Response from the server
"""
event_data = {
'event_type': event_type,
'content': content,
'metadata': metadata or {},
'timestamp': datetime.now().isoformat(),
'event_id': f"evt_{int(time.time() * 1000)}"
}
print(f"\n{'='*80}")
print(f"📤 SENDING EVENT")
print(f"{'='*80}")
print(f"Event Type: {event_type}")
print(f"Content: {content}")
if metadata:
print(f"Metadata: {json.dumps(metadata, indent=2)}")
print(f"{'='*80}\n")
try:
response = requests.post(
f"{self.server_url}/event",
json=event_data,
headers={'Content-Type': 'application/json'},
timeout=120
)
response.raise_for_status()
result = response.json()
print(f"\n{'='*80}")
print(f"✅ EVENT SENT SUCCESSFULLY")
print(f"{'='*80}")
print(f"Response: {json.dumps(result, indent=2)}")
print(f"{'='*80}\n")
return result
except requests.exceptions.RequestException as e:
print(f"\n❌ Error sending event: {e}")
return {"error": str(e)}
def reset_agent(self) -> dict:
"""Reset the agent state"""
try:
response = requests.post(f"{self.server_url}/agent/reset")
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
return {"error": str(e)}
def get_status(self) -> dict:
"""Get agent status"""
try:
response = requests.get(f"{self.server_url}/agent/status")
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
return {"error": str(e)}
def start_monitoring(self) -> dict:
"""Start system monitoring"""
try:
response = requests.post(f"{self.server_url}/monitoring/start")
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
return {"error": str(e)}
def stop_monitoring(self) -> dict:
"""Stop system monitoring"""
try:
response = requests.post(f"{self.server_url}/monitoring/stop")
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
return {"error": str(e)}
def register_process(self, process_id: str, name: str) -> dict:
"""Register a background process for monitoring"""
try:
response = requests.post(
f"{self.server_url}/process/register",
json={'process_id': process_id, 'name': name}
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
return {"error": str(e)}
def unregister_process(self, process_id: str) -> dict:
"""Unregister a background process"""
try:
response = requests.post(
f"{self.server_url}/process/unregister",
json={'process_id': process_id}
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
return {"error": str(e)}
def run_test_scenarios(client: EventClient):
"""Run various test scenarios"""
print("\n" + "🧪"*40)
print(" EVENT-TRIGGERED AGENT TEST SCENARIOS")
print("🧪"*40 + "\n")
# Scenario 1: Web message
print("\n📋 Scenario 1: Web Interface Message")
print("-"*80)
client.send_event(
event_type=EventType.WEB_MESSAGE.value,
content="Hello! Can you create a simple Python script that prints 'Hello, World!'?",
metadata={"user_id": "user123", "session_id": "session456"}
)
time.sleep(2)
# Scenario 2: IM message
print("\n📋 Scenario 2: Instant Message")
print("-"*80)
client.send_event(
event_type=EventType.IM_MESSAGE.value,
content="Can you list the files in the current directory?",
metadata={"sender": "Alice", "platform": "Slack"}
)
time.sleep(2)
# Scenario 3: Email reply
print("\n📋 Scenario 3: Email Reply")
print("-"*80)
client.send_event(
event_type=EventType.EMAIL_REPLY.value,
content="Thanks for the report! Can you also check the disk usage?",
metadata={
"from": "bob@example.com",
"subject": "Re: System Report",
"thread_id": "thread789"
}
)
time.sleep(2)
# Scenario 4: GitHub PR update
print("\n📋 Scenario 4: GitHub PR Review")
print("-"*80)
client.send_event(
event_type=EventType.GITHUB_PR_UPDATE.value,
content="Review comment: Please add unit tests for the new feature.",
metadata={
"pr_number": "42",
"action": "review_requested",
"reviewer": "code-reviewer",
"repository": "ai-agent-project"
}
)
time.sleep(2)
# Scenario 5: Timer trigger
print("\n📋 Scenario 5: Scheduled Timer")
print("-"*80)
client.send_event(
event_type=EventType.TIMER_TRIGGER.value,
content="Daily backup reminder - please check if backups are running correctly.",
metadata={
"timer_id": "daily_backup_check",
"schedule": "daily at 09:00"
}
)
time.sleep(2)
# Scenario 6: System alert
print("\n📋 Scenario 6: System Alert")
print("-"*80)
client.send_event(
event_type=EventType.SYSTEM_ALERT.value,
content="Memory usage has exceeded 80%. Please investigate.",
metadata={
"alert_type": "resource_usage",
"severity": "warning",
"memory_usage": "82%"
}
)
time.sleep(2)
# Scenario 7: Register background process
print("\n📋 Scenario 7: Background Process Registration")
print("-"*80)
print("Registering background process...")
result = client.register_process("proc_ml_training", "ML Model Training")
print(f"Result: {json.dumps(result, indent=2)}")
# Scenario 8: Start monitoring
print("\n📋 Scenario 8: Start System Monitoring")
print("-"*80)
print("Starting system monitoring (will check for timeouts)...")
result = client.start_monitoring()
print(f"Result: {json.dumps(result, indent=2)}")
print("\n⏰ Monitoring is now active. System will check for:")
print(" - User timeout (no interaction for 1 minute)")
print(" - Background process timeout (running for 30 seconds)")
print("\n💡 Wait 1-2 minutes to see system reminder events trigger automatically...")
# Get status
print("\n📋 Current Agent Status")
print("-"*80)
status = client.get_status()
print(json.dumps(status, indent=2))
print("\n" + "✅"*40)
print(" TEST SCENARIOS COMPLETED")
print("✅"*40 + "\n")
def interactive_mode(client: EventClient):
"""Interactive mode for sending custom events"""
print("\n" + "="*80)
print(" INTERACTIVE EVENT CLIENT")
print("="*80)
print("\nAvailable event types:")
for event_type in EventType:
print(f" - {event_type.value}")
print("\nCommands:")
print(" 'status' - Get agent status")
print(" 'reset' - Reset agent")
print(" 'monitor on' - Start monitoring")
print(" 'monitor off' - Stop monitoring")
print(" 'quit' - Exit")
print("\nOr send an event: <event_type> <content>")
while True:
try:
print("\n" + "-"*60)
user_input = input("Event > ").strip()
if not user_input:
continue
if user_input.lower() == 'quit':
print("👋 Goodbye!")
break
elif user_input.lower() == 'status':
status = client.get_status()
print(json.dumps(status, indent=2))
elif user_input.lower() == 'reset':
result = client.reset_agent()
print(json.dumps(result, indent=2))
elif user_input.lower() == 'monitor on':
result = client.start_monitoring()
print(json.dumps(result, indent=2))
elif user_input.lower() == 'monitor off':
result = client.stop_monitoring()
print(json.dumps(result, indent=2))
else:
# Parse event command
parts = user_input.split(' ', 1)
if len(parts) < 2:
print("❌ Invalid format. Use: <event_type> <content>")
continue
event_type = parts[0]
content = parts[1]
# Validate event type
try:
EventType(event_type)
except ValueError:
print(f"❌ Invalid event type: {event_type}")
continue
# Send the event
client.send_event(event_type, content)
except KeyboardInterrupt:
print("\n\n⚠️ Interrupted. Type 'quit' to exit.")
except Exception as e:
print(f"\n❌ Error: {str(e)}")
def main():
"""Main entry point"""
parser = argparse.ArgumentParser(
description="事件客户端：向事件驱动 Agent 服务器发送事件。",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""示例：
python client.py --mode test # 依次发送多种事件，跑通全部场景
python client.py --mode interactive # 交互模式，手动输入事件
python client.py --message "创建一个 hello world 脚本" # 发送单条 web_message 事件
python client.py --event-type timer_trigger --message "检查每日备份" # 指定事件类型
""",
)
parser.add_argument(
'--server',
default='http://localhost:8000',
help='服务器地址（默认：http://localhost:8000）'
)
parser.add_argument(
'--mode',
choices=['test', 'interactive'],
default='test',
help='模式：test（依次发送预置场景事件）或 interactive（交互式手动发送）'
)
parser.add_argument(
'--message',
default=None,
help='发送单条事件的内容；提供该参数时忽略 --mode，发完即退出'
)
parser.add_argument(
'--event-type',
default=EventType.WEB_MESSAGE.value,
choices=[e.value for e in EventType],
help=f'--message 使用的事件类型（默认：{EventType.WEB_MESSAGE.value}）'
)
args = parser.parse_args()
client = EventClient(server_url=args.server)
# Check if server is running
try:
response = requests.get(f"{args.server}/health", timeout=5)
response.raise_for_status()
print(f"✅ Connected to server at {args.server}")
except requests.exceptions.RequestException as e:
print(f"❌ Cannot connect to server at {args.server}")
print(f" Error: {e}")
print(f"\n💡 Make sure the server is running:")
print(f" python server.py")
return
if args.message is not None:
client.send_event(event_type=args.event_type, content=args.message)
elif args.mode == 'test':
run_test_scenarios(client)
else:
interactive_mode(client)
if __name__ == "__main__":
main()
event_loop_demo.py¶
"""
event_loop_demo.py —— 事件驱动 Agent 的端到端演示（单进程、可离线运行）
本章"事件驱动的异步 Agent"一节指出：真正的"主动服务"不仅需要 Agent 能定时
检查世界，更需要世界能主动通知 Agent。本脚本用最小的代码把这一点跑起来——
1. 注册若干"事件触发器"（trigger source），每个触发器在后台线程里运行，
在事件真正发生的那一刻把一个结构化 Event 推入统一的事件队列：
- 一次性定时器 OneShotTimer —— 对应书中 set_timer 的"一次性定时器"
- 循环定时器 RecurringTimer —— 对应书中 set_timer 的"循环定时器"
- 文件监听 FileWatchTrigger —— 对应 n8n 等平台的文件变更触发器
2. 事件循环 EventLoop 从队列里逐个取出事件，唤醒 Agent 处理——这正是
"Agent 注册、外部触发"的完整闭环：注册时声明关心什么事件，触发时被异步唤醒。
与需要起 HTTP 服务器的 server.py / client.py 不同，本脚本在单个进程里同时扮演
"外部世界"和"Agent"，因此适合用来直观演示事件驱动的行为。
离线模式（--mock）：不调用大模型，用一个"模拟动作"打印 Agent 被唤醒后的处理
过程，可在没有 API Key 的环境下观察完整的触发→唤醒→处理闭环。
真实模式（默认）：接入 EventTriggeredAgent，由大模型真正处理每个事件。
用法示例：
python event_loop_demo.py --mock # 离线演示全部触发器
python event_loop_demo.py --mock --trigger timer # 只演示一次性定时器
python event_loop_demo.py --mock --trigger recurring --interval 3 --duration 12
python event_loop_demo.py --trigger file --watch-dir ./watched # 真实 Agent 处理文件事件
"""
import os
import sys
import time
import queue
import logging
import argparse
import threading
from datetime import datetime
from typing import Optional, Callable
from event_types import Event, EventType
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger("event_loop")
# ============================================================================
# 事件触发器（trigger source）
# ============================================================================
class TriggerSource(threading.Thread):
"""事件触发器基类：在后台线程中运行，把事件推入共享的事件队列。
注册（register）体现在实例化并 start()；触发（fire）体现在 run() 中
满足条件时调用 self.emit(event)。这与书中"注册时由 Agent 主动调用工具、
触发时由外部事件异步回调"的两个时刻一一对应。
"""
def __init__(self, name: str, event_queue: "queue.Queue[Event]"):
super().__init__(name=name, daemon=True)
self.event_queue = event_queue
self._stop = threading.Event()
def emit(self, event: Event):
"""触发：把事件推入事件队列，唤醒事件循环。"""
logger.info(f"⚡ [{self.name}] 触发事件 -> {event.event_type.value}: {event.content}")
self.event_queue.put(event)
def stop(self):
self._stop.set()
class OneShotTimer(TriggerSource):
"""一次性定时器：延迟 delay 秒后触发一次 timer_trigger 事件。
对应书中"用户要求给 DMV 打电话，当前是周六，Agent 设置'下周一上午 10:00
致电 DMV'"这类有明确时间点的任务。
"""
def __init__(self, event_queue, delay: float, content: str, timer_id: str = "oneshot"):
super().__init__(name=f"OneShotTimer({timer_id})", event_queue=event_queue)
self.delay = delay
self.content = content
self.timer_id = timer_id
def run(self):
logger.info(f"⏱️ [{self.name}] 已注册：{self.delay:.0f} 秒后触发")
if self._stop.wait(self.delay):
return
self.emit(Event(
event_type=EventType.TIMER_TRIGGER,
content=self.content,
metadata={"timer_id": self.timer_id, "kind": "one_shot",
"scheduled_delay_seconds": self.delay},
))
class RecurringTimer(TriggerSource):
"""循环定时器：每隔 interval 秒触发一次 timer_trigger 事件。
对应书中"每小时检查一次服务器健康状况""每周五发送进展报告"，以及
OpenClaw Heartbeat 式的定时轮询。
"""
def __init__(self, event_queue, interval: float, content: str, timer_id: str = "recurring"):
super().__init__(name=f"RecurringTimer({timer_id})", event_queue=event_queue)
self.interval = interval
self.content = content
self.timer_id = timer_id
def run(self):
logger.info(f"🔁 [{self.name}] 已注册：每 {self.interval:.0f} 秒触发一次")
tick = 0
while not self._stop.wait(self.interval):
tick += 1
self.emit(Event(
event_type=EventType.TIMER_TRIGGER,
content=f"{self.content}（第 {tick} 次）",
metadata={"timer_id": self.timer_id, "kind": "recurring",
"interval_seconds": self.interval, "tick": tick},
))
class FileWatchTrigger(TriggerSource):
"""文件监听：轮询目录，发现新增或被修改的文件时触发 file_change 事件。
对应书中"n8n 等工作流平台的触发器生态：Webhook、定时器、邮件、数据库
变更、文件监听"。这里用轮询实现，不依赖第三方库，便于跨平台离线运行。
"""
def __init__(self, event_queue, watch_dir: str, poll_interval: float = 1.0):
super().__init__(name=f"FileWatch({watch_dir})", event_queue=event_queue)
self.watch_dir = watch_dir
self.poll_interval = poll_interval
self._snapshot = {}
def _scan(self):
snapshot = {}
try:
for entry in os.scandir(self.watch_dir):
if entry.is_file():
snapshot[entry.name] = entry.stat().st_mtime
except FileNotFoundError:
pass
return snapshot
def run(self):
os.makedirs(self.watch_dir, exist_ok=True)
self._snapshot = self._scan()
logger.info(f"👀 [{self.name}] 已注册：轮询间隔 {self.poll_interval:.0f} 秒"
f"（当前已有 {len(self._snapshot)} 个文件）")
while not self._stop.wait(self.poll_interval):
current = self._scan()
for name, mtime in current.items():
if name not in self._snapshot:
change = "created"
elif mtime != self._snapshot[name]:
change = "modified"
else:
continue
self.emit(Event(
event_type=EventType.FILE_CHANGE,
content=f"检测到文件{'新增' if change == 'created' else '修改'}，请查看其内容并给出简要处理建议。",
metadata={"path": os.path.join(self.watch_dir, name), "change": change},
))
self._snapshot = current
# ============================================================================
# 事件循环（event loop）
# ============================================================================
class EventLoop:
"""统一事件队列 + 单线程分发。
所有触发器把异构事件推入同一个队列；事件循环按到达顺序取出，每个事件
唤醒一次 Agent 处理。这正是书中"将所有输入统一建模为事件流，通过事件
循环驱动 Agent 的思考和行动"的最小实现。
"""
def __init__(self, dispatch: Callable[[Event], None]):
self.event_queue: "queue.Queue[Event]" = queue.Queue()
self.dispatch = dispatch
self.triggers = []
self.processed = 0
def add_trigger(self, trigger: TriggerSource):
self.triggers.append(trigger)
def run(self, duration: float):
"""启动所有触发器，运行 duration 秒后停止。"""
deadline = time.monotonic() + duration
for t in self.triggers:
t.start()
logger.info(f"🟢 事件循环启动，将运行 {duration:.0f} 秒，等待事件唤醒 Agent...\n")
while time.monotonic() < deadline:
try:
event = self.event_queue.get(timeout=0.5)
except queue.Empty:
continue
self.processed += 1
logger.info(f"\n{'='*80}\n📥 事件循环取出第 {self.processed} 个事件"
f" -> 唤醒 Agent\n{'='*80}")
try:
self.dispatch(event)
except Exception as e: # noqa: BLE001 - 演示中不希望单个事件异常终止循环
logger.error(f"❌ 处理事件时出错: {e}")
for t in self.triggers:
t.stop()
logger.info(f"\n🔴 事件循环结束，共处理 {self.processed} 个事件。")
# ============================================================================
# 分发处理器：模拟动作 or 真实 Agent
# ============================================================================
def make_mock_dispatch() -> Callable[[Event], None]:
"""离线模拟处理器：不调用大模型，打印 Agent 被唤醒后的处理过程。"""
def dispatch(event: Event):
logger.info(f"🤖 Agent 被唤醒，收到消息: {event.to_user_message()}")
# 用一个确定性的"模拟动作"代替大模型 + 工具调用
if event.event_type == EventType.TIMER_TRIGGER:
action = "读取定时任务上下文 -> 执行例行检查 -> 汇报结果"
elif event.event_type == EventType.FILE_CHANGE:
path = event.metadata.get("path", "")
preview = ""
try:
with open(path, "r", encoding="utf-8", errors="replace") as f:
preview = f.read(120).replace("\n", " ")
except OSError:
preview = "(无法读取文件内容)"
action = f"读取文件 {os.path.basename(path)} -> 内容预览: {preview!r} -> 生成处理建议"
else:
action = "解析事件 -> 调用相关工具 -> 生成处理结果"
logger.info(f"🛠️ [模拟动作] {action}")
logger.info(f"✅ Agent 处理完成: 已响应 {event.event_type.value} 事件")
return dispatch
def make_agent_dispatch(provider: str, model: Optional[str],
max_iterations: int) -> Callable[[Event], None]:
"""真实处理器：接入 EventTriggeredAgent，由大模型处理每个事件。"""
from agent import EventTriggeredAgent, SystemHintConfig, resolve_provider_and_key
# 通用兜底：直连 provider 的 key 缺失时，若有 OPENROUTER_API_KEY 则自动改走 openrouter。
resolved_provider, api_key = resolve_provider_and_key(provider)
if not api_key:
print(f"❌ 未检测到 provider '{provider}' 对应的 API Key（也未配置 OPENROUTER_API_KEY 兜底）。")
print(f" 请先设置环境变量，或改用离线演示：python event_loop_demo.py --mock")
sys.exit(1)
if resolved_provider != provider:
print(f"ℹ️ provider '{provider}' 无可用 Key，已自动改用 OpenRouter 兜底（openrouter）。")
provider = resolved_provider
# 保留已是 provider/model 形式的显式模型；否则让 openrouter 用其默认模型。
model = model if (model and "/" in model) else None
config = SystemHintConfig(
enable_timestamps=True,
enable_tool_counter=True,
enable_todo_list=True,
enable_detailed_errors=True,
enable_system_state=True,
save_trajectory=True,
trajectory_file="event_loop_trajectory.json",
temperature=0.7,
max_tokens=4096,
use_mcp_servers=False, # 本演示仅用内置工具，避免额外的 MCP 依赖
)
agent = EventTriggeredAgent(api_key=api_key, provider=provider,
model=model, config=config, verbose=True)
logger.info(f"✅ 真实 Agent 初始化完成（provider={provider}, model={agent.model}）")
def dispatch(event: Event):
result = agent.handle_event(event, max_iterations=max_iterations)
logger.info(f"✅ Agent 处理完成: success={result['success']}, "
f"iterations={result['iterations']}, "
f"tool_calls={len(result['tool_calls'])}")
return dispatch
# ============================================================================
# CLI
# ============================================================================
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="事件驱动 Agent 端到端演示：注册触发器，由外部事件异步唤醒 Agent。",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""示例：
python event_loop_demo.py --mock
离线演示全部触发器（一次性定时器 + 循环定时器 + 文件监听），无需 API Key
python event_loop_demo.py --mock --trigger timer
只演示一次性定时器
python event_loop_demo.py --mock --trigger recurring --interval 3 --duration 12
每 3 秒触发一次循环定时器，共运行 12 秒
python event_loop_demo.py --mock --trigger file --watch-dir ./watched
监听 ./watched 目录，向其中写入文件即可触发事件
python event_loop_demo.py --trigger timer --provider kimi
用真实大模型处理一次性定时器事件（需要设置对应的 API Key）
""",
)
parser.add_argument(
"--trigger", choices=["timer", "recurring", "file", "all"], default="all",
help="要演示的触发器类型：timer=一次性定时器，recurring=循环定时器，"
"file=文件监听，all=全部（默认：all）",
)
parser.add_argument(
"--mock", action="store_true",
help="离线模式：不调用大模型，用模拟动作演示触发→唤醒→处理闭环（无需 API Key）",
)
parser.add_argument(
"--duration", type=float, default=12.0,
help="事件循环总运行时长（秒），到时后停止所有触发器（默认：12）",
)
parser.add_argument(
"--delay", type=float, default=3.0,
help="一次性定时器的延迟触发时间（秒）（默认：3）",
)
parser.add_argument(
"--interval", type=float, default=4.0,
help="循环定时器的触发间隔（秒）（默认：4）",
)
parser.add_argument(
"--watch-dir", default="watched_dir",
help="文件监听触发器监视的目录，不存在会自动创建（默认：watched_dir）",
)
parser.add_argument(
"--provider", default=os.getenv("LLM_PROVIDER", "kimi"),
choices=["siliconflow", "doubao", "kimi", "moonshot", "openrouter"],
help="真实模式使用的大模型提供商（默认：环境变量 LLM_PROVIDER 或 kimi）",
)
parser.add_argument(
"--model", default=os.getenv("LLM_MODEL"),
help="真实模式的模型名覆盖（默认：使用提供商默认模型）",
)
parser.add_argument(
"--max-iterations", type=int, default=10,
help="真实模式下单个事件的最大工具调用轮数（默认：10）",
)
return parser
def main():
args = build_parser().parse_args()
print("\n" + "=" * 80)
print("🚀 事件驱动 Agent 演示（EVENT-DRIVEN AGENT DEMO）")
print("=" * 80)
print(f"触发器: {args.trigger} | 模式: {'离线模拟' if args.mock else '真实 Agent'} | "
f"时长: {args.duration:.0f}s")
print("=" * 80 + "\n")
sys.stdout.flush()
if args.mock:
dispatch = make_mock_dispatch()
else:
dispatch = make_agent_dispatch(args.provider, args.model, args.max_iterations)
loop = EventLoop(dispatch)
if args.trigger in ("timer", "all"):
loop.add_trigger(OneShotTimer(
loop.event_queue, delay=args.delay, timer_id="daily_backup_check",
content="一次性定时器到期：请检查每日备份是否已经完成。",
))
if args.trigger in ("recurring", "all"):
loop.add_trigger(RecurringTimer(
loop.event_queue, interval=args.interval, timer_id="health_check",
content="循环定时器到期：请检查服务器健康状况。",
))
if args.trigger in ("file", "all"):
loop.add_trigger(FileWatchTrigger(loop.event_queue, watch_dir=args.watch_dir))
print(f"💡 提示：向目录 {args.watch_dir}/ 写入或修改文件即可触发 file_change 事件。")
print(f" 例如另开一个终端执行：echo hello > {args.watch_dir}/note.txt\n")
sys.stdout.flush()
if not loop.triggers:
print("❌ 没有可运行的触发器。")
sys.exit(1)
try:
loop.run(duration=args.duration)
except KeyboardInterrupt:
print("\n⚠️ 收到中断信号，正在停止...")
for t in loop.triggers:
t.stop()
print("\n" + "=" * 80)
print(f"📊 演示结束：共处理 {loop.processed} 个事件。")
print("=" * 80 + "\n")
if __name__ == "__main__":
main()
event_types.py¶
"""
Event types for the event-triggered agent system
"""
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Dict, Any, Optional
class EventType(Enum):
"""Types of events that can trigger agent actions"""
# External input events
WEB_MESSAGE = "web_message"
IM_MESSAGE = "im_message"
EMAIL_REPLY = "email_reply"
GITHUB_PR_UPDATE = "github_pr_update"
TIMER_TRIGGER = "timer_trigger"
FILE_CHANGE = "file_change"
# System reminder events
USER_TIMEOUT = "user_timeout"
PROCESS_TIMEOUT = "process_timeout"
SYSTEM_ALERT = "system_alert"
@dataclass
class Event:
"""Represents an event that triggers agent action"""
event_type: EventType
content: str
metadata: Dict[str, Any] = field(default_factory=dict)
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
event_id: Optional[str] = None
def to_user_message(self) -> str:
"""Convert event to user message format for the agent"""
if self.event_type == EventType.WEB_MESSAGE:
return f"[Web Interface] {self.content}"
elif self.event_type == EventType.IM_MESSAGE:
sender = self.metadata.get('sender', 'Unknown')
return f"[IM from {sender}] {self.content}"
elif self.event_type == EventType.EMAIL_REPLY:
from_email = self.metadata.get('from', 'Unknown')
subject = self.metadata.get('subject', 'No Subject')
return f"[Email Reply from {from_email}]\nSubject: {subject}\n{self.content}"
elif self.event_type == EventType.GITHUB_PR_UPDATE:
pr_number = self.metadata.get('pr_number', 'Unknown')
action = self.metadata.get('action', 'updated')
return f"[GitHub PR #{pr_number} {action}] {self.content}"
elif self.event_type == EventType.TIMER_TRIGGER:
timer_id = self.metadata.get('timer_id', 'Unknown')
return f"[Timer {timer_id} triggered] {self.content}"
elif self.event_type == EventType.FILE_CHANGE:
path = self.metadata.get('path', 'Unknown')
change = self.metadata.get('change', 'modified')
return f"[File {change}: {path}] {self.content}"
elif self.event_type == EventType.USER_TIMEOUT:
duration = self.metadata.get('duration', 'unknown')
return f"[System Reminder] User has not responded for {duration}. {self.content}"
elif self.event_type == EventType.PROCESS_TIMEOUT:
process_id = self.metadata.get('process_id', 'Unknown')
duration = self.metadata.get('duration', 'unknown')
return f"[System Alert] Background process {process_id} has been running for {duration}. {self.content}"
elif self.event_type == EventType.SYSTEM_ALERT:
alert_type = self.metadata.get('alert_type', 'general')
return f"[System Alert: {alert_type}] {self.content}"
else:
return f"[{self.event_type.value}] {self.content}"
def to_dict(self) -> Dict[str, Any]:
"""Convert event to dictionary for JSON serialization"""
return {
'event_type': self.event_type.value,
'content': self.content,
'metadata': self.metadata,
'timestamp': self.timestamp,
'event_id': self.event_id
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'Event':
"""Create event from dictionary"""
return cls(
event_type=EventType(data['event_type']),
content=data['content'],
metadata=data.get('metadata', {}),
timestamp=data.get('timestamp', datetime.now().isoformat()),
event_id=data.get('event_id')
)
example_with_mcp.py¶
"""
Example demonstrating the Event-Triggered Agent with MCP tools
"""
import os
import asyncio
from dotenv import load_dotenv
from agent import EventTriggeredAgent, SystemHintConfig
from event_types import Event, EventType
# Load environment variables
load_dotenv()
async def main():
"""Main example function"""
print("=" * 80)
print("Event-Triggered Agent with MCP Tools Example")
print("=" * 80)
print()
# Get API credentials
provider = os.getenv("LLM_PROVIDER", "kimi")
api_key = os.getenv("KIMI_API_KEY")
if not api_key:
print("❌ Please set KIMI_API_KEY in your .env file")
return
# Create agent configuration
config = SystemHintConfig(
enable_timestamps=True,
enable_tool_counter=True,
enable_todo_list=True,
enable_detailed_errors=True,
enable_system_state=True,
save_trajectory=True,
trajectory_file="example_trajectory.json",
use_mcp_servers=True # Enable MCP servers
)
# Initialize agent
print("Initializing agent...")
agent = EventTriggeredAgent(
api_key=api_key,
provider=provider,
config=config,
verbose=True
)
# Load MCP tools
print("\nLoading MCP tools...")
await agent.load_mcp_tools()
print("\n" + "=" * 80)
print("Testing Event Processing")
print("=" * 80)
print()
# Create a test event
event = Event(
event_type=EventType.WEB_MESSAGE,
content="Search the web for 'Python async programming best practices' and summarize the top 3 results.",
metadata={
"source": "web_interface",
"user_id": "demo_user",
"session_id": "test_session_001"
}
)
# Handle the event
try:
result = agent.handle_event(event, max_iterations=15)
print("\n" + "=" * 80)
print("Result Summary")
print("=" * 80)
print(f"Success: {result['success']}")
print(f"Iterations: {result['iterations']}")
print(f"Tool Calls: {len(result['tool_calls'])}")
if result.get('final_answer'):
print(f"\nFinal Answer:\n{result['final_answer']}")
if result.get('trajectory_file'):
print(f"\nTrajectory saved to: {result['trajectory_file']}")
except Exception as e:
print(f"\n❌ Error processing event: {e}")
import traceback
traceback.print_exc()
finally:
# Cleanup MCP connections
print("\nCleaning up MCP connections...")
await agent.mcp_manager.disconnect_all()
print("✅ Cleanup complete")
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\n\n⚠️ Interrupted by user")
quickstart.py¶
"""
Quick Start Script for Event-Triggered Agent
Demonstrates the basic functionality in a simple way
"""
import os
import sys
import time
import subprocess
import signal
from event_types import EventType
# Check if API key is set (universal OpenRouter fallback applied by the server).
from agent import resolve_provider_and_key
provider = os.getenv("LLM_PROVIDER", "kimi").lower()
resolved_provider, api_key = resolve_provider_and_key(provider)
if not api_key:
print(f"❌ Error: no API key for provider '{provider}', and no OPENROUTER_API_KEY fallback")
print(f"\nPlease set one of:")
print(f" export KIMI_API_KEY='...' # or SILICONFLOW/DOUBAO/OPENROUTER per provider")
print(f" export OPENROUTER_API_KEY='...' # universal fallback")
print(f"\nOr change provider:")
print(f" export LLM_PROVIDER=kimi # or siliconflow, doubao, openrouter")
sys.exit(1)
if resolved_provider != provider:
print(f"ℹ️ provider '{provider}' has no key; the server will fall back to OpenRouter.")
print("\n" + "="*80)
print("🚀 EVENT-TRIGGERED AGENT QUICK START")
print("="*80)
print()
# Check if server is already running
import requests
try:
response = requests.get("http://localhost:8000/health", timeout=2)
print("✅ Server is already running!")
print("\n💡 You can now use the client to send events:")
print(" python client.py --mode test")
print(" python client.py --mode interactive")
sys.exit(0)
except:
pass
print("📦 Starting the event-triggered agent server...")
print("\n⏳ This may take a moment to initialize...\n")
# Start the server in a subprocess
try:
server_process = subprocess.Popen(
[sys.executable, "server.py"],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
universal_newlines=True,
bufsize=1
)
# Wait for server to start
print("⏰ Waiting for server to start...")
max_wait = 30
for i in range(max_wait):
try:
response = requests.get("http://localhost:8000/health", timeout=1)
if response.status_code == 200:
print("✅ Server is running!\n")
break
except:
pass
time.sleep(1)
if i % 5 == 0:
print(f" Still waiting... ({i}/{max_wait}s)")
else:
print("❌ Server failed to start in time")
server_process.terminate()
sys.exit(1)
print("="*80)
print("🎉 QUICK START READY!")
print("="*80)
print()
print("The event-triggered agent server is now running on port 8000.")
print()
print("📋 What you can do now:")
print()
print("1. Send test events (in another terminal):")
print(" python client.py --mode test")
print()
print("2. Use interactive mode:")
print(" python client.py --mode interactive")
print()
print("3. Send individual events via API:")
print(" curl -X POST http://localhost:8000/event \\")
print(" -H 'Content-Type: application/json' \\")
print(" -d '{\"event_type\": \"web_message\", \"content\": \"Hello!\"}'")
print()
print("4. Check agent status:")
print(" curl http://localhost:8000/agent/status")
print()
print("="*80)
print("📺 Server output will appear below:")
print("="*80)
print()
# Stream server output
try:
while True:
line = server_process.stdout.readline()
if not line:
break
print(line, end='')
except KeyboardInterrupt:
print("\n\n⚠️ Shutting down server...")
server_process.send_signal(signal.SIGINT)
server_process.wait(timeout=5)
print("✅ Server stopped")
except FileNotFoundError:
print("❌ Error: Could not find server.py")
print("Make sure you're in the agent-with-event-trigger directory")
sys.exit(1)
except Exception as e:
print(f"❌ Error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
server.py¶
"""
Event Server - FastAPI version with native async support for MCP tools
"""
import os
import logging
from datetime import datetime, timedelta
from typing import Dict, Any, Optional
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from agent import EventTriggeredAgent, SystemHintConfig, resolve_provider_and_key
from event_types import Event, EventType
import threading
import time
import asyncio
import argparse
import uvicorn
def _env_int(name: str, default: int) -> int:
"""Read an integer env var; fall back to default (with a warning) if malformed."""
raw = os.getenv(name)
if raw is None:
return default
try:
return int(raw)
except ValueError:
logger.warning(f"Invalid {name} value: {raw!r} (must be an integer); using default {default}")
return default
def _reasoning_safe_temperature(model, requested=1.0):
"""Reasoning models (Kimi K3, GPT-5, ...) only accept temperature=1.
Return 1 for those; otherwise the requested value so non-reasoning
providers (Doubao, DeepSeek, older Moonshot) are unchanged."""
m = str(model or "").lower().replace("/", "-")
return 1 if ("kimi-k3" in m or "gpt-5" in m) else requested
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# Global agent instance
agent: Optional[EventTriggeredAgent] = None
agent_lock = threading.Lock()
# Monitoring state
monitoring_enabled = False
monitoring_thread: Optional[threading.Thread] = None
# MCP loading status
mcp_loading_status = {
"loading": False,
"loaded": False,
"tools_count": 0,
"error": None,
"started_at": None,
"completed_at": None
}
# ============================================================================
# FastAPI Lifecycle Events (Modern lifespan)
# ============================================================================
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Lifespan context manager for startup and shutdown"""
global agent, monitoring_enabled
# Startup
logger.info("🚀 Starting Event-Triggered Agent Server (FastAPI)")
await init_agent()
logger.info("✅ Server ready to receive events\n")
yield
# Shutdown
logger.info("Shutting down server...")
monitoring_enabled = False
if agent and agent.mcp_manager:
await agent.mcp_manager.disconnect_all()
logger.info("✅ Server shutdown complete")
# Initialize FastAPI app with lifespan
app = FastAPI(
title="Event-Triggered Agent Server",
description="AI Agent with async MCP tools support",
version="2.0.0",
lifespan=lifespan
)
# Pydantic models for requests
class EventRequest(BaseModel):
event_type: str
content: str
metadata: Optional[Dict[str, Any]] = None
class ProcessRegister(BaseModel):
process_id: str
process_name: str
metadata: Optional[Dict[str, Any]] = None
class ProcessUnregister(BaseModel):
process_id: str
# ============================================================================
# Initialization
# ============================================================================
async def init_agent():
"""Initialize the agent with optional MCP tools"""
global agent, mcp_loading_status
# Determine provider from environment (universal OpenRouter fallback applied)
requested_provider = os.getenv("LLM_PROVIDER", "kimi").lower()
provider, api_key = resolve_provider_and_key(requested_provider)
if not api_key:
raise ValueError(
f"API key not set for provider '{requested_provider}'. Set the appropriate "
f"environment variable, or set OPENROUTER_API_KEY as a universal fallback."
)
# Get model from environment if specified
model = os.getenv("LLM_MODEL")
if provider != requested_provider:
logger.info(
f"ℹ️ provider '{requested_provider}' has no key; falling back to OpenRouter."
)
# Keep an explicit provider/model id; otherwise use OpenRouter's default.
if not (model and "/" in model):
model = None
# Check if MCP should be enabled (default: true)
enable_mcp = os.getenv("ENABLE_MCP_TOOLS", "true").lower() not in ["false", "0", "no"]
config = SystemHintConfig(
enable_timestamps=True,
enable_tool_counter=True,
enable_todo_list=True,
enable_detailed_errors=True,
enable_system_state=True,
save_trajectory=True,
trajectory_file="event_agent_trajectory.json",
temperature=_reasoning_safe_temperature(model, 0.7),
max_tokens=4096,
use_mcp_servers=enable_mcp
)
agent = EventTriggeredAgent(
api_key=api_key,
provider=provider,
model=model,
config=config,
verbose=True
)
logger.info(f"✅ Agent initialized with {provider} provider")
if enable_mcp:
logger.info("🔄 MCP tools enabled (default) - loading asynchronously...")
await load_mcp_tools_async()
else:
logger.info(f"📦 Using built-in tools only (MCP disabled via ENABLE_MCP_TOOLS=false)")
async def load_mcp_tools_async():
"""Load MCP tools asynchronously"""
global agent, mcp_loading_status
mcp_loading_status["loading"] = True
mcp_loading_status["started_at"] = datetime.now().isoformat()
try:
if agent:
await agent.load_mcp_tools()
tools_count = len(agent.mcp_manager.tools)
mcp_loading_status["loaded"] = True
mcp_loading_status["loading"] = False
mcp_loading_status["tools_count"] = tools_count
mcp_loading_status["completed_at"] = datetime.now().isoformat()
logger.info(f"✅ MCP tools loaded: {tools_count} tools available")
if tools_count > 0:
sample_tools = list(agent.mcp_manager.tools.keys())[:5]
logger.info(f" Sample: {sample_tools}")
else:
raise RuntimeError("Agent not initialized")
except Exception as e:
logger.error(f"❌ Failed to load MCP tools: {e}")
mcp_loading_status["loading"] = False
mcp_loading_status["loaded"] = False
mcp_loading_status["error"] = str(e)
mcp_loading_status["completed_at"] = datetime.now().isoformat()
# ============================================================================
# API Endpoints
# ============================================================================
@app.get("/")
async def root():
"""Root endpoint with API information"""
return {
"service": "Event-Triggered Agent Server",
"version": "2.0.0",
"status": "running",
"docs": "/docs",
"endpoints": {
"health": "GET /health",
"event": "POST /event",
"mcp_status": "GET /mcp/status",
"mcp_reload": "POST /mcp/reload",
"agent_status": "GET /agent/status",
"agent_reset": "POST /agent/reset"
}
}
@app.get("/health")
async def health_check():
"""Health check endpoint"""
return {
"status": "healthy",
"agent_initialized": agent is not None,
"monitoring_enabled": monitoring_enabled,
"mcp_enabled": agent.config.use_mcp_servers if agent else False,
"mcp_loaded": mcp_loading_status["loaded"],
"timestamp": datetime.now().isoformat()
}
@app.get("/mcp/status")
async def get_mcp_status():
"""Get MCP tools loading status"""
status = mcp_loading_status.copy()
# Add tool list if loaded
if status["loaded"] and agent:
status["tools"] = list(agent.mcp_manager.tools.keys())
# Group by server
status["tools_by_server"] = {}
for tool_name in agent.mcp_manager.tools.keys():
server = tool_name.split("_")[0]
if server not in status["tools_by_server"]:
status["tools_by_server"][server] = []
status["tools_by_server"][server].append(tool_name)
return status
@app.post("/mcp/reload")
async def reload_mcp_tools(background_tasks: BackgroundTasks):
"""Manually trigger MCP tools reload"""
if agent is None:
raise HTTPException(status_code=500, detail="Agent not initialized")
if mcp_loading_status["loading"]:
raise HTTPException(status_code=409, detail="MCP tools are already loading")
# Use FastAPI background tasks
background_tasks.add_task(load_mcp_tools_async)
return {
"success": True,
"message": "MCP tools reload started in background"
}
@app.post("/event")
async def handle_event(event_req: EventRequest):
"""Handle incoming event"""
if agent is None:
raise HTTPException(status_code=500, detail="Agent not initialized")
try:
# Create event
event_data = {
"event_type": event_req.event_type,
"content": event_req.content,
"metadata": event_req.metadata or {}
}
event = Event.from_dict(event_data)
# Handle the event
with agent_lock:
result = agent.handle_event(event, max_iterations=20)
return {
"success": True,
"event_id": event.event_id,
"result": {
"final_answer": result.get('final_answer'),
"iterations": result.get('iterations'),
"tool_calls_count": len(result.get('tool_calls', [])),
"todo_items": len(result.get('todo_list', [])),
"success": result.get('success', False),
"trajectory_file": result.get('trajectory_file')
}
}
except Exception as e:
logger.error(f"Error handling event: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/agent/status")
async def get_agent_status():
"""Get current agent status"""
if agent is None:
raise HTTPException(status_code=500, detail="Agent not initialized")
with agent_lock:
return {
"provider": agent.provider,
"model": agent.model,
"tool_calls_count": len(agent.tool_calls),
"todo_items": len(agent.todo_list),
"current_directory": agent.current_directory,
"mcp_tools_loaded": agent.mcp_tools_loaded,
"mcp_tools_count": len(agent.mcp_manager.tools) if agent.mcp_tools_loaded else 0
}
@app.post("/agent/reset")
async def reset_agent():
"""Reset agent state"""
if agent is None:
raise HTTPException(status_code=500, detail="Agent not initialized")
with agent_lock:
agent.reset()
return {
"success": True,
"message": "Agent state reset successfully"
}
@app.post("/process/register")
async def register_process(process: ProcessRegister):
"""Register a background process for monitoring"""
if agent is None:
raise HTTPException(status_code=500, detail="Agent not initialized")
with agent_lock:
agent.background_processes[process.process_id] = {
"name": process.process_name,
"start_time": datetime.now().isoformat(),
"metadata": process.metadata or {},
"reminded": False
}
return {
"success": True,
"message": f"Process '{process.process_name}' registered"
}
@app.post("/process/unregister")
async def unregister_process(process: ProcessUnregister):
"""Unregister a background process"""
if agent is None:
raise HTTPException(status_code=500, detail="Agent not initialized")
with agent_lock:
if process.process_id in agent.background_processes:
del agent.background_processes[process.process_id]
return {
"success": True,
"message": f"Process {process.process_id} unregistered"
}
else:
return {
"success": False,
"message": f"Process {process.process_id} not found"
}
# ============================================================================
# Main Entry Point
# ============================================================================
def build_parser() -> argparse.ArgumentParser:
"""构建命令行参数解析器（命令行参数优先级高于环境变量）。"""
parser = argparse.ArgumentParser(
description="事件驱动 Agent 的 HTTP 服务器（FastAPI）："
"对外暴露 /event 等接口，把 Webhook 式的外部回调转成事件唤醒 Agent。",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""示例：
python server.py # 使用默认配置（端口 8000，启用 MCP 工具）
python server.py --port 9000 # 自定义端口
python server.py --provider doubao # 指定大模型提供商
python server.py --no-mcp # 只用内置工具，不加载 MCP 工具
之后用客户端发送事件：python client.py --mode test
""",
)
parser.add_argument(
"--host", default=os.getenv("AGENT_HOST", "0.0.0.0"),
help="监听地址（默认：0.0.0.0）",
)
parser.add_argument(
"--port", type=int, default=_env_int("AGENT_PORT", 8000),
help="监听端口（默认：环境变量 AGENT_PORT 或 8000）",
)
parser.add_argument(
"--provider", default=None,
choices=["siliconflow", "doubao", "kimi", "moonshot", "openrouter"],
help="大模型提供商（默认：环境变量 LLM_PROVIDER 或 kimi）",
)
parser.add_argument(
"--model", default=None,
help="模型名覆盖（默认：使用提供商默认模型）",
)
parser.add_argument(
"--no-mcp", action="store_true",
help="禁用 MCP 工具，只使用内置工具（等价于 ENABLE_MCP_TOOLS=false）",
)
return parser
def main():
"""Main entry point"""
args = build_parser().parse_args()
# 命令行参数覆盖环境变量：init_agent() 在 lifespan 中读取这些环境变量
if args.provider:
os.environ["LLM_PROVIDER"] = args.provider
if args.model:
os.environ["LLM_MODEL"] = args.model
if args.no_mcp:
os.environ["ENABLE_MCP_TOOLS"] = "false"
print("\n" + "="*80)
print("🤖 EVENT-TRIGGERED AGENT SERVER (FastAPI)")
print("="*80)
print()
print(f"✅ Starting server on {args.host}:{args.port}")
print(f"📡 API Documentation: http://localhost:{args.port}/docs")
print(f"📊 ReDoc: http://localhost:{args.port}/redoc")
print()
print("="*80 + "\n")
# Run with uvicorn
uvicorn.run(
app,
host=args.host,
port=args.port,
log_level="info"
)
if __name__ == "__main__":
main()
server_fastapi.py¶
"""
Event Server - FastAPI version with native async support for MCP tools
"""
import os
import logging
from datetime import datetime, timedelta
from typing import Dict, Any, Optional
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from agent import EventTriggeredAgent, SystemHintConfig
from event_types import Event, EventType
import threading
import time
import asyncio
import uvicorn
def _env_int(name: str, default: int) -> int:
"""Read an integer env var; fall back to default (with a warning) if malformed."""
raw = os.getenv(name)
if raw is None:
return default
try:
return int(raw)
except ValueError:
logger.warning(f"Invalid {name} value: {raw!r} (must be an integer); using default {default}")
return default
def _reasoning_safe_temperature(model, requested=1.0):
"""Reasoning models (Kimi K3, GPT-5, ...) only accept temperature=1.
Return 1 for those; otherwise the requested value so non-reasoning
providers (Doubao, DeepSeek, older Moonshot) are unchanged."""
m = str(model or "").lower().replace("/", "-")
return 1 if ("kimi-k3" in m or "gpt-5" in m) else requested
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# Global agent instance
agent: Optional[EventTriggeredAgent] = None
agent_lock = threading.Lock()
# Monitoring state
monitoring_enabled = False
monitoring_thread: Optional[threading.Thread] = None
# MCP loading status
mcp_loading_status = {
"loading": False,
"loaded": False,
"tools_count": 0,
"error": None,
"started_at": None,
"completed_at": None
}
# ============================================================================
# FastAPI Lifecycle Events (Modern lifespan)
# ============================================================================
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Lifespan context manager for startup and shutdown"""
global agent, monitoring_enabled
# Startup
logger.info("🚀 Starting Event-Triggered Agent Server (FastAPI)")
await init_agent()
logger.info("✅ Server ready to receive events\n")
yield
# Shutdown
logger.info("Shutting down server...")
monitoring_enabled = False
if agent and agent.mcp_manager:
await agent.mcp_manager.disconnect_all()
logger.info("✅ Server shutdown complete")
# Initialize FastAPI app with lifespan
app = FastAPI(
title="Event-Triggered Agent Server",
description="AI Agent with async MCP tools support",
version="2.0.0",
lifespan=lifespan
)
# Pydantic models for requests
class EventRequest(BaseModel):
event_type: str
content: str
metadata: Optional[Dict[str, Any]] = None
class ProcessRegister(BaseModel):
process_id: str
process_name: str
metadata: Optional[Dict[str, Any]] = None
class ProcessUnregister(BaseModel):
process_id: str
# ============================================================================
# Initialization
# ============================================================================
async def init_agent():
"""Initialize the agent with optional MCP tools"""
global agent, mcp_loading_status
# Determine provider from environment
provider = os.getenv("LLM_PROVIDER", "kimi").lower()
# Get API key based on provider
if provider == "siliconflow":
api_key = os.getenv("SILICONFLOW_API_KEY")
elif provider == "doubao":
api_key = os.getenv("DOUBAO_API_KEY")
elif provider in ["kimi", "moonshot"]:
api_key = os.getenv("KIMI_API_KEY")
elif provider == "openrouter":
api_key = os.getenv("OPENROUTER_API_KEY")
else:
raise ValueError(f"Unsupported provider: {provider}")
if not api_key:
raise ValueError(f"API key not set for provider '{provider}'. Set the appropriate environment variable.")
# Get model from environment if specified
model = os.getenv("LLM_MODEL")
# Check if MCP should be enabled (default: true)
enable_mcp = os.getenv("ENABLE_MCP_TOOLS", "true").lower() not in ["false", "0", "no"]
config = SystemHintConfig(
enable_timestamps=True,
enable_tool_counter=True,
enable_todo_list=True,
enable_detailed_errors=True,
enable_system_state=True,
save_trajectory=True,
trajectory_file="event_agent_trajectory.json",
temperature=_reasoning_safe_temperature(model, 0.7),
max_tokens=4096,
use_mcp_servers=enable_mcp
)
agent = EventTriggeredAgent(
api_key=api_key,
provider=provider,
model=model,
config=config,
verbose=True
)
logger.info(f"✅ Agent initialized with {provider} provider")
if enable_mcp:
logger.info("🔄 MCP tools enabled (default) - loading asynchronously...")
await load_mcp_tools_async()
else:
logger.info(f"📦 Using built-in tools only (MCP disabled via ENABLE_MCP_TOOLS=false)")
async def load_mcp_tools_async():
"""Load MCP tools asynchronously"""
global agent, mcp_loading_status
mcp_loading_status["loading"] = True
mcp_loading_status["started_at"] = datetime.now().isoformat()
try:
if agent:
await agent.load_mcp_tools()
tools_count = len(agent.mcp_manager.tools)
mcp_loading_status["loaded"] = True
mcp_loading_status["loading"] = False
mcp_loading_status["tools_count"] = tools_count
mcp_loading_status["completed_at"] = datetime.now().isoformat()
logger.info(f"✅ MCP tools loaded: {tools_count} tools available")
if tools_count > 0:
sample_tools = list(agent.mcp_manager.tools.keys())[:5]
logger.info(f" Sample: {sample_tools}")
else:
raise RuntimeError("Agent not initialized")
except Exception as e:
logger.error(f"❌ Failed to load MCP tools: {e}")
mcp_loading_status["loading"] = False
mcp_loading_status["loaded"] = False
mcp_loading_status["error"] = str(e)
mcp_loading_status["completed_at"] = datetime.now().isoformat()
# ============================================================================
# API Endpoints
# ============================================================================
@app.get("/")
async def root():
"""Root endpoint with API information"""
return {
"service": "Event-Triggered Agent Server",
"version": "2.0.0",
"status": "running",
"docs": "/docs",
"endpoints": {
"health": "GET /health",
"event": "POST /event",
"mcp_status": "GET /mcp/status",
"mcp_reload": "POST /mcp/reload",
"agent_status": "GET /agent/status",
"agent_reset": "POST /agent/reset"
}
}
@app.get("/health")
async def health_check():
"""Health check endpoint"""
return {
"status": "healthy",
"agent_initialized": agent is not None,
"monitoring_enabled": monitoring_enabled,
"mcp_enabled": agent.config.use_mcp_servers if agent else False,
"mcp_loaded": mcp_loading_status["loaded"],
"timestamp": datetime.now().isoformat()
}
@app.get("/mcp/status")
async def get_mcp_status():
"""Get MCP tools loading status"""
status = mcp_loading_status.copy()
# Add tool list if loaded
if status["loaded"] and agent:
status["tools"] = list(agent.mcp_manager.tools.keys())
# Group by server
status["tools_by_server"] = {}
for tool_name in agent.mcp_manager.tools.keys():
server = tool_name.split("_")[0]
if server not in status["tools_by_server"]:
status["tools_by_server"][server] = []
status["tools_by_server"][server].append(tool_name)
return status
@app.post("/mcp/reload")
async def reload_mcp_tools(background_tasks: BackgroundTasks):
"""Manually trigger MCP tools reload"""
if agent is None:
raise HTTPException(status_code=500, detail="Agent not initialized")
if mcp_loading_status["loading"]:
raise HTTPException(status_code=409, detail="MCP tools are already loading")
# Use FastAPI background tasks
background_tasks.add_task(load_mcp_tools_async)
return {
"success": True,
"message": "MCP tools reload started in background"
}
@app.post("/event")
async def handle_event(event_req: EventRequest):
"""Handle incoming event"""
if agent is None:
raise HTTPException(status_code=500, detail="Agent not initialized")
try:
# Create event
event_data = {
"event_type": event_req.event_type,
"content": event_req.content,
"metadata": event_req.metadata or {}
}
event = Event.from_dict(event_data)
# Handle the event
with agent_lock:
result = agent.handle_event(event, max_iterations=20)
return {
"success": True,
"event_id": event.event_id,
"result": {
"final_answer": result.get('final_answer'),
"iterations": result.get('iterations'),
"tool_calls_count": len(result.get('tool_calls', [])),
"todo_items": len(result.get('todo_list', [])),
"success": result.get('success', False),
"trajectory_file": result.get('trajectory_file')
}
}
except Exception as e:
logger.error(f"Error handling event: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/agent/status")
async def get_agent_status():
"""Get current agent status"""
if agent is None:
raise HTTPException(status_code=500, detail="Agent not initialized")
with agent_lock:
return {
"provider": agent.provider,
"model": agent.model,
"tool_calls_count": len(agent.tool_calls),
"todo_items": len(agent.todo_list),
"current_directory": agent.current_directory,
"mcp_tools_loaded": agent.mcp_tools_loaded,
"mcp_tools_count": len(agent.mcp_manager.tools) if agent.mcp_tools_loaded else 0
}
@app.post("/agent/reset")
async def reset_agent():
"""Reset agent state"""
if agent is None:
raise HTTPException(status_code=500, detail="Agent not initialized")
with agent_lock:
agent.reset()
return {
"success": True,
"message": "Agent state reset successfully"
}
@app.post("/process/register")
async def register_process(process: ProcessRegister):
"""Register a background process for monitoring"""
if agent is None:
raise HTTPException(status_code=500, detail="Agent not initialized")
with agent_lock:
agent.background_processes[process.process_id] = {
"name": process.process_name,
"start_time": datetime.now().isoformat(),
"metadata": process.metadata or {},
"reminded": False
}
return {
"success": True,
"message": f"Process '{process.process_name}' registered"
}
@app.post("/process/unregister")
async def unregister_process(process: ProcessUnregister):
"""Unregister a background process"""
if agent is None:
raise HTTPException(status_code=500, detail="Agent not initialized")
with agent_lock:
if process.process_id in agent.background_processes:
del agent.background_processes[process.process_id]
return {
"success": True,
"message": f"Process {process.process_id} unregistered"
}
else:
return {
"success": False,
"message": f"Process {process.process_id} not found"
}
# ============================================================================
# Main Entry Point
# ============================================================================
def main():
"""Main entry point"""
print("\n" + "="*80)
print("🤖 EVENT-TRIGGERED AGENT SERVER (FastAPI)")
print("="*80)
print()
# Get port from environment
port = _env_int('AGENT_PORT', 8000)
print(f"✅ Starting server on port {port}")
print(f"📡 API Documentation: http://localhost:{port}/docs")
print(f"📊 ReDoc: http://localhost:{port}/redoc")
print()
print("="*80 + "\n")
# Run with uvicorn
uvicorn.run(
app,
host="0.0.0.0",
port=port,
log_level="info"
)
if __name__ == "__main__":
main()
test_demo.py¶
"""
Simple demo script to test the event-triggered agent locally
without needing the server/client architecture
"""
import os
from agent import EventTriggeredAgent, SystemHintConfig
from event_types import Event, EventType
def _reasoning_safe_temperature(model, requested=1.0):
"""Reasoning models (Kimi K3, GPT-5, ...) only accept temperature=1.
Return 1 for those; otherwise the requested value so non-reasoning
providers (Doubao, DeepSeek, older Moonshot) are unchanged."""
m = str(model or "").lower().replace("/", "-")
return 1 if ("kimi-k3" in m or "gpt-5" in m) else requested
def main():
"""Run a simple demo of the event-triggered agent"""
print("\n" + "="*80)
print("🧪 EVENT-TRIGGERED AGENT DEMO")
print("="*80)
print()
# Get provider and API key (matching conversational_agent.py)
provider = os.getenv("LLM_PROVIDER", "kimi").lower()
if provider == "siliconflow":
api_key = os.getenv("SILICONFLOW_API_KEY")
elif provider == "doubao":
api_key = os.getenv("DOUBAO_API_KEY")
elif provider in ["kimi", "moonshot"]:
api_key = os.getenv("KIMI_API_KEY")
elif provider == "openrouter":
api_key = os.getenv("OPENROUTER_API_KEY")
else:
print(f"❌ Error: Unsupported provider: {provider}")
return
if not api_key:
print(f"❌ Error: Please set API key for provider '{provider}'")
print(f" export {provider.upper()}_API_KEY='your-api-key-here'")
return
# Get optional model override
model = os.getenv("LLM_MODEL")
# Create agent with full system hints (matching conversational_agent.py config)
config = SystemHintConfig(
enable_timestamps=True,
enable_tool_counter=True,
enable_todo_list=True,
enable_detailed_errors=True,
enable_system_state=True,
save_trajectory=True,
trajectory_file="demo_trajectory.json",
temperature=_reasoning_safe_temperature(model, 0.7), # Matching conversational_agent.py
max_tokens=4096 # Matching conversational_agent.py
)
agent = EventTriggeredAgent(
api_key=api_key,
provider=provider,
model=model,
config=config,
verbose=True
)
print("✅ Agent initialized\n")
# Demo 1: Web message
print("\n" + "-"*80)
print("📋 Demo 1: Web Interface Message")
print("-"*80)
event1 = Event(
event_type=EventType.WEB_MESSAGE,
content="Create a simple Python script that prints 'Hello, Event-Triggered Agent!' and save it as demo_hello.py",
metadata={"user_id": "demo_user"}
)
result1 = agent.handle_event(event1, max_iterations=10)
print(f"\n✅ Event handled. Success: {result1['success']}")
print(f" Iterations: {result1['iterations']}")
print(f" Tool calls: {len(result1['tool_calls'])}")
# Demo 2: IM message
print("\n" + "-"*80)
print("📋 Demo 2: Instant Message")
print("-"*80)
event2 = Event(
event_type=EventType.IM_MESSAGE,
content="Can you run the script you just created?",
metadata={"sender": "Alice", "platform": "Slack"}
)
result2 = agent.handle_event(event2, max_iterations=10)
print(f"\n✅ Event handled. Success: {result2['success']}")
print(f" Iterations: {result2['iterations']}")
print(f" Tool calls: {len(result2['tool_calls'])}")
# Demo 3: System alert
print("\n" + "-"*80)
print("📋 Demo 3: System Alert")
print("-"*80)
event3 = Event(
event_type=EventType.SYSTEM_ALERT,
content="Please check the current directory and list all Python files.",
metadata={"alert_type": "routine_check"}
)
result3 = agent.handle_event(event3, max_iterations=10)
print(f"\n✅ Event handled. Success: {result3['success']}")
print(f" Iterations: {result3['iterations']}")
print(f" Tool calls: {len(result3['tool_calls'])}")
# Summary
print("\n" + "="*80)
print("📊 DEMO SUMMARY")
print("="*80)
print(f"Total events processed: 3")
print(f"Total tool calls: {len(agent.tool_calls)}")
print(f"Total conversation messages: {len(agent.conversation_history)}")
print(f"Trajectory saved to: {config.trajectory_file}")
print()
print("✅ Demo completed successfully!")
print()
print("💡 Next steps:")
print(" - Check demo_hello.py to see the created file")
print(" - View demo_trajectory.json to see the full conversation")
print(" - Run 'python server.py' and 'python client.py' for the full system")
print("="*80 + "\n")
if __name__ == "__main__":
main()
test_env_int.py¶
"""Regression test: malformed AGENT_PORT must not crash server startup.
Both server variants parsed AGENT_PORT with bare int() (one inside
build_parser's default, one in main), so AGENT_PORT=abc crashed with an
unhandled ValueError at startup. They now fall back to 8000 with a warning.
"""
import os
import sys
sys.path.insert(0, os.path.dirname(__file__))
import server
import server_fastapi
def test_env_int_falls_back_on_malformed(monkeypatch):
monkeypatch.setenv("AGENT_PORT", "abc")
assert server._env_int("AGENT_PORT", 8000) == 8000
assert server_fastapi._env_int("AGENT_PORT", 8000) == 8000
def test_env_int_parses_valid_value(monkeypatch):
monkeypatch.setenv("AGENT_PORT", "9000")
assert server._env_int("AGENT_PORT", 8000) == 9000
assert server_fastapi._env_int("AGENT_PORT", 8000) == 9000
def test_env_int_default_when_unset(monkeypatch):
monkeypatch.delenv("AGENT_PORT", raising=False)
assert server._env_int("AGENT_PORT", 8000) == 8000
def test_build_parser_survives_malformed_env(monkeypatch):
monkeypatch.setenv("AGENT_PORT", "not-a-port")
args = server.build_parser().parse_args([])
assert args.port == 8000