1+ """Asynchronous resource implementation for long-running operations."""
2+
3+ import asyncio
4+ import enum
5+ # import time
6+ from typing import Any , Optional
7+
8+ import pydantic
9+ from pydantic import Field
10+
11+ from mcp .server .fastmcp .resources .base import Resource
12+ from mcp .server .fastmcp .utilities .logging import get_logger
13+
14+ logger = get_logger (__name__ )
15+
16+
17+ class AsyncStatus (str , enum .Enum ):
18+ """Status of an asynchronous operation."""
19+ PENDING = "pending"
20+ RUNNING = "running"
21+ COMPLETED = "completed"
22+ FAILED = "failed"
23+ CANCELED = "canceled"
24+
25+
26+ class AsyncResource (Resource ):
27+ """A resource representing an asynchronous operation.
28+
29+ This resource type is used to track long-running operations that are executed
30+ asynchronously. It provides methods for updating progress, completing with a result,
31+ failing with an error, and canceling the operation.
32+ """
33+
34+ status : AsyncStatus = Field (
35+ default = AsyncStatus .PENDING ,
36+ description = "Current status of the asynchronous operation"
37+ )
38+ # progress: float = Field(
39+ # default=0,
40+ # description="Current progress value (0-100 or raw count)"
41+ # )
42+ error : Optional [str ] = Field (
43+ default = None ,
44+ description = "Error message if the operation failed"
45+ )
46+ # created_at: float = Field(
47+ # default_factory=time.time,
48+ # description="Timestamp when the resource was created"
49+ # )
50+ # started_at: Optional[float] = Field(
51+ # default=None,
52+ # description="Timestamp when the operation started running"
53+ # )
54+ # completed_at: Optional[float] = Field(
55+ # default=None,
56+ # description="Timestamp when the operation completed, failed, or was canceled"
57+ # )
58+
59+ # Fields not included in serialization
60+ _task : Optional [asyncio .Task [Any ]] = pydantic .PrivateAttr (default = None )
61+ # _mcp_server = pydantic.PrivateAttr(default=None)
62+
63+ # def set_mcp_server(self, server: Any) -> None:
64+ # """Set the MCP server reference.
65+
66+ # Args:
67+ # server: The MCP server instance
68+ # """
69+ # self._mcp_server = server
70+
71+ async def read (self ) -> str :
72+ """Read the current state of the resource as JSON.
73+
74+ Returns the current status and progress information.
75+ """
76+ # Convert the resource to a dictionary, excluding private fields
77+ data = self .model_dump (exclude = {"_task" })
78+
79+ # Return status info as JSON
80+ import json
81+ return json .dumps (data , indent = 2 )
82+
83+ async def start (self , task : asyncio .Task [Any ]) -> None :
84+ """Mark the resource as running and store the task.
85+
86+ Args:
87+ task: The asyncio task that is executing the operation
88+ """
89+ self ._task = task
90+ self .status = AsyncStatus .RUNNING
91+ # self.started_at = time.time()
92+ # await self._notify_changed()
93+
94+ logger .debug (
95+ "Started async operation" ,
96+ extra = {
97+ "uri" : self .uri ,
98+ }
99+ )
100+
101+ # async def update_progress(self, progress: float) -> None:
102+ # """Update the progress information.
103+
104+ # Args:
105+ # progress: Current progress value
106+ # total: Total expected progress value, if known
107+ # """
108+ # self.progress = progress
109+ # # await self._notify_changed()
110+
111+ # logger.debug(
112+ # "Updated async operation progress",
113+ # extra={
114+ # "uri": self.uri,
115+ # "progress": self.progress,
116+ # }
117+ # )
118+
119+ async def complete (self ) -> None :
120+ """Mark the resource as completed.
121+ """
122+ self .status = AsyncStatus .COMPLETED
123+ # self.completed_at = time.time()
124+
125+ # await self._notify_changed()
126+
127+ logger .info (
128+ "Completed async operation" ,
129+ extra = {
130+ "uri" : self .uri ,
131+ # "duration": self.completed_at - (self.started_at or self.created_at),
132+ }
133+ )
134+
135+ async def fail (self , error : str ) -> None :
136+ """Mark the resource as failed and store the error.
137+
138+ Args:
139+ error: Error message describing why the operation failed
140+ """
141+ self .status = AsyncStatus .FAILED
142+ self .error = error
143+ # self.completed_at = time.time()
144+ # await self._notify_changed()
145+
146+ logger .error (
147+ "Failed async operation" ,
148+ extra = {
149+ "uri" : self .uri ,
150+ "error" : error ,
151+ # "duration": self.completed_at - (self.started_at or self.created_at),
152+ }
153+ )
154+
155+ async def cancel (self ) -> None :
156+ """Cancel the operation if it's still running."""
157+ if self .status in (AsyncStatus .PENDING , AsyncStatus .RUNNING ) and self ._task :
158+ self ._task .cancel ()
159+ self .status = AsyncStatus .CANCELED
160+ # self.completed_at = time.time()
161+ # await self._notify_changed()
162+
163+ logger .info (
164+ "Canceled async operation" ,
165+ extra = {
166+ "uri" : self .uri ,
167+ # "duration": self.completed_at - (self.started_at or self.created_at),
168+ }
169+ )
170+
171+ # async def _notify_changed(self) -> None:
172+ # """Notify subscribers that the resource has changed."""
173+ # if self._mcp_server:
174+ # # This will be implemented in the MCP server to notify clients
175+ # # of resource changes via the notification protocol
176+ # self._mcp_server.notify_resource_changed(self.uri)
0 commit comments