from dataclasses import dataclassfrom typing import List, Optionalfrom datetime import datetime@dataclassclass TodoItem: id: int title: str completed: bool = False created_at: datetime = None def __post_init__(self): if self.created_at is None: self.created_at = datetime.now()class TodoApp: def __init__(self): self.todos: List[TodoItem] = [] self.next_id = 1 def add_todo(self, title: str) -> TodoItem: """Add a new TODO item""" todo = TodoItem(id=self.next_id, title=title) self.todos.append(todo) self.next_id += 1 return todo def complete_todo(self, todo_id: int) -> bool: """Mark a TODO as completed""" for todo in self.todos: if todo.id == todo_id: todo.completed = True return True return False def delete_todo(self, todo_id: int) -> bool: """Delete a TODO item""" for i, todo in enumerate(self.todos): if todo.id == todo_id: self.todos.pop(i) return True return False def list_todos(self) -> List[dict]: """Get all TODOs as dictionaries""" return [ { "id": todo.id, "title": todo.title, "completed": todo.completed, "created_at": todo.created_at.isoformat() } for todo in self.todos ] def clear_all(self): """Reset the TODO list""" self.todos = [] self.next_id = 1
import sysimport loggingfrom hud.server import MCPServerfrom .todo_app import TodoApp# Configure logging to stderrlogging.basicConfig( stream=sys.stderr, level=logging.INFO, format='[%(levelname)s] %(asctime)s | %(name)s | %(message)s')# Create server and app instancesmcp = MCPServer(name="TODO Environment")app = TodoApp()# Interaction tools (visible to agents)@mcp.tool()async def add_todo(title: str) -> dict: """Add a new TODO item to the list""" todo = app.add_todo(title) return { "success": True, "todo": { "id": todo.id, "title": todo.title }, "message": f"Added TODO: {title}" }@mcp.tool()async def complete_todo(todo_id: int) -> dict: """Mark a TODO item as completed""" success = app.complete_todo(todo_id) return { "success": success, "message": f"{'Completed' if success else 'Failed to complete'} TODO {todo_id}" }@mcp.tool()async def list_todos() -> dict: """List all TODO items""" todos = app.list_todos() return { "todos": todos, "total": len(todos), "completed": sum(1 for t in todos if t["completed"]) }@mcp.tool()async def delete_todo(todo_id: int) -> dict: """Delete a TODO item""" success = app.delete_todo(todo_id) return { "success": success, "message": f"{'Deleted' if success else 'Failed to delete'} TODO {todo_id}" }# Setup tool (hidden from agents)@mcp.tool()async def setup(initial_todos: list[str] = None) -> dict: """Initialize TODO list with optional items""" app.clear_all() if initial_todos: for title in initial_todos: app.add_todo(title) return { "status": "success", "message": f"Initialized with {len(initial_todos or [])} TODOs" }# Evaluation tool (hidden from agents)@mcp.tool()async def evaluate(target_completed: int = None) -> dict: """Evaluate based on TODO completion""" todos = app.list_todos() total = len(todos) completed = sum(1 for t in todos if t["completed"]) if target_completed is not None: # Specific target reward = 1.0 if completed >= target_completed else completed / target_completed else: # General completion rate reward = completed / total if total > 0 else 0.0 return { "reward": reward, "done": reward >= 1.0, "info": { "completed": completed, "total": total, "details": f"{completed}/{total} TODOs completed" } }# Initialize handler@mcp.initializeasync def on_initialize(ctx): """Called when MCP connection is established""" logging.info("TODO environment ready!")if __name__ == "__main__": mcp.run()