subscribe_resource
Subscribe to Taiwan Stock Agent MCP server resource updates to cache and receive notifications for real-time stock data, historical trends, and market analysis.
Instructions
Subscribe to resource updates for caching and notifications
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| resource_uri | Yes |
Implementation Reference
- mcp_server.py:276-304 (handler)The main handler function for the 'subscribe_resource' tool, registered via @mcp.tool decorator. It calls the resource_manager to subscribe and returns subscription status with metadata.@mcp.tool(name="subscribe_resource", description="Subscribe to resource updates for caching and notifications") async def subscribe_resource_tool(resource_uri: str) -> Dict[str, Any]: """Subscribe to resource updates.""" try: success = resource_manager.subscribe_to_resource(resource_uri) return { "resource_uri": resource_uri, "subscribed": success, "active_subscriptions": resource_manager.get_subscriptions(), "_metadata": { "source": "tw-stock-agent", "timestamp": datetime.now().isoformat(), "data_type": "resource_subscription" } } except Exception as e: logger.error(f"Failed to subscribe to resource {resource_uri}: {e}") return { "resource_uri": resource_uri, "subscribed": False, "error": str(e), "_metadata": { "source": "tw-stock-agent", "timestamp": datetime.now().isoformat(), "data_type": "resource_subscription", "has_error": True } }
- The core implementation of resource subscription in ResourceManager class, which adds the resource URI to the subscriptions set.def subscribe_to_resource(self, resource_uri: str) -> bool: """ Subscribe to resource updates. Args: resource_uri: The resource URI to subscribe to Returns: True if subscription successful """ try: self.subscriptions.add(resource_uri) logger.info(f"Subscribed to resource: {resource_uri}") return True except Exception as e: logger.error(f"Failed to subscribe to resource {resource_uri}: {e}") return False