1  #!/usr/bin/env python3
 2  """
 3  MCP Tasks API Example
 4  
 5  Demonstrates the Tasks API per MCP 2025-11-25 specification.
 6  Tasks are durable state machines for tracking long-running operations.
 7  
 8  Usage:
 9      python mcp_tasks_example.py
10  """
11  
12  import asyncio
13  from praisonai.mcp_server.tasks import (
14      TaskManager,
15      TaskStatus,
16  )
17  
18  
19  async def simulate_long_operation(method: str, params: dict) -> str:
20      """Simulate a long-running operation."""
21      print(f"  Executing: {method} with {params}")
22      await asyncio.sleep(1)  # Simulate work
23      return f"Result for {method}: Success!"
24  
25  
26  async def main():
27      print("=" * 60)
28      print("MCP Tasks API Example (2025-11-25 Specification)")
29      print("=" * 60)
30      
31      # Create task manager with executor
32      manager = TaskManager(executor=simulate_long_operation)
33      
34      # 1. Create a task
35      print("\n1. Creating a task...")
36      task = await manager.create_task(
37          method="tools/call",
38          params={"name": "search", "arguments": {"query": "AI news"}},
39          metadata={"user": "demo"},
40          execute=True,  # Start execution immediately
41      )
42      print(f"   Task created: {task.id}")
43      print(f"   Status: {task.status.value}")
44      print(f"   Created at: {task.created_at}")
45      
46      # 2. Get task status (poll)
47      print("\n2. Polling task status...")
48      for i in range(3):
49          await asyncio.sleep(0.5)
50          current = manager.get_task(task.id)
51          if current:
52              print(f"   Poll {i+1}: status={current.status.value}")
53              if current.status in (TaskStatus.COMPLETED, TaskStatus.FAILED):
54                  break
55      
56      # 3. Get final result
57      print("\n3. Getting task result...")
58      final = manager.get_task(task.id)
59      if final:
60          print(f"   Final status: {final.status.value}")
61          print(f"   Result: {final.result}")
62          
63          # Show MCP-compliant response format
64          print("\n   MCP Response Format:")
65          task_dict = final.to_dict()
66          for key, value in task_dict.items():
67              print(f"     {key}: {value}")
68      
69      # 4. Create another task and cancel it
70      print("\n4. Creating and cancelling a task...")
71      task2 = await manager.create_task(
72          method="tools/call",
73          params={"name": "slow_operation"},
74          execute=False,  # Don't execute yet
75      )
76      print(f"   Task created: {task2.id}")
77      
78      cancelled = await manager.cancel_task(task2.id)
79      if cancelled:
80          print(f"   Task cancelled: {cancelled.status.value}")
81          print(f"   Status message: {cancelled.status_message}")
82      
83      # 5. List all tasks
84      print("\n5. Listing all tasks...")
85      tasks = manager.list_tasks()
86      for t in tasks:
87          print(f"   - {t.id}: {t.status.value}")
88      
89      print("\n" + "=" * 60)
90      print("Tasks API Example Complete!")
91      print("=" * 60)
92  
93  
94  if __name__ == "__main__":
95      asyncio.run(main())