feat: added simple resources and better prompts [2025-08-01]

Author: CHRISCARLON

src/api_service/os_api.py

@@ -283,7 +283,6 @@ async def make_request(
                     timeout=timeout,
                 ) as response:
                     if response.status >= 400:
-                        # Sanitize error response text
                         error_text = await response.text()
                         sanitized_error = self._sanitise_api_key(error_text)
                         error_message = (
@@ -311,3 +310,72 @@ async def make_request(
         raise RuntimeError(
             "Unreachable: make_request exited retry loop without returning or raising"
         )
+
+    async def make_request_no_auth(
+        self,
+        url: str,
+        params: Optional[Dict[str, Any]] = None,
+        max_retries: int = 2,
+    ) -> str:
+        """
+        Make a request without authentication (for public endpoints like documentation).
+
+        Args:
+            url: Full URL to request
+            params: Additional query parameters
+            max_retries: Maximum number of retries for transient errors
+
+        Returns:
+            Response text (not JSON parsed)
+        """
+        await self.initialise()
+
+        if self.session is None:
+            raise ValueError("Session not initialised")
+
+        current_time = asyncio.get_event_loop().time()
+        elapsed = current_time - self.last_request_time
+        if elapsed < self.request_delay:
+            await asyncio.sleep(self.request_delay - elapsed)
+
+        request_params = params or {}
+        headers = {"User-Agent": self.user_agent}
+
+        logger.info(f"Requesting URL (no auth): {url}")
+
+        for attempt in range(1, max_retries + 1):
+            try:
+                self.last_request_time = asyncio.get_event_loop().time()
+
+                timeout = aiohttp.ClientTimeout(total=30.0)
+                async with self.session.get(
+                    url,
+                    params=request_params,
+                    headers=headers,
+                    timeout=timeout,
+                ) as response:
+                    if response.status >= 400:
+                        error_text = await response.text()
+                        error_message = f"HTTP Error: {response.status} - {error_text}"
+                        logger.error(f"Error: {error_message}")
+                        raise ValueError(error_message)
+
+                    return await response.text()
+
+            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
+                if attempt == max_retries:
+                    error_message = (
+                        f"Request failed after {max_retries} attempts: {str(e)}"
+                    )
+                    logger.error(f"Error: {error_message}")
+                    raise ValueError(error_message)
+                else:
+                    await asyncio.sleep(0.7)
+            except Exception as e:
+                error_message = f"Request failed: {str(e)}"
+                logger.error(f"Error: {error_message}")
+                raise ValueError(error_message)
+
+        raise RuntimeError(
+            "Unreachable: make_request_no_auth exited retry loop without returning or raising"
+        )

src/api_service/protocols.py

@@ -26,6 +26,15 @@ async def make_request(
         """Make a request to an API endpoint"""
         ...
 
+    async def make_request_no_auth(
+        self,
+        url: str,
+        params: Optional[Dict[str, Any]] = None,
+        max_retries: int = 2,
+    ) -> str:
+        """Make a request without authentication"""
+        ...
+
     async def cache_openapi_spec(self):
         """Cache the OpenAPI spec"""
         ...

src/mcp_service/os_service.py

@@ -13,6 +13,8 @@
 from workflow_generator.workflow_planner import WorkflowPlanner
 from utils.logging_config import get_logger
 from models import Collection
+from mcp_service.resources import OSDocumentationResources
+from mcp_service.prompts import OSWorkflowPrompts
 
 logger = get_logger(__name__)
 
@@ -37,6 +39,13 @@ def __init__(
         self.workflow_planner: Optional[WorkflowPlanner] = None
         self.guardrails = ToolGuardrails()
         self.register_tools()
+        self.register_resources()
+        self.register_prompts()
+
+    def register_resources(self) -> None:
+        """Register all MCP resources"""
+        doc_resources = OSDocumentationResources(self.mcp, self.api_client)
+        doc_resources.register_all()
 
     def register_tools(self) -> None:
         """Register all MCP tools with guardrails and middleware"""
@@ -76,6 +85,11 @@ def apply_middleware(func):
             apply_middleware(self.get_prompt_templates)
         )
 
+    def register_prompts(self) -> None:
+        """Register all MCP prompts"""
+        workflow_prompts = OSWorkflowPrompts(self.mcp)
+        workflow_prompts.register_all()
+
     def run(self) -> None:
         """Run the MCP service"""
         try:

src/mcp_service/prompts.py

@@ -0,0 +1,72 @@
+from typing import List
+from mcp.types import PromptMessage, TextContent
+from prompt_templates.prompt_templates import PROMPT_TEMPLATES
+from utils.logging_config import get_logger
+
+logger = get_logger(__name__)
+
+
+class OSWorkflowPrompts:
+    """Handles registration of OS NGD workflow prompts"""
+    
+    def __init__(self, mcp_service):
+        self.mcp = mcp_service
+    
+    def register_all(self) -> None:
+        """Register all workflow prompts"""
+        self._register_analysis_prompts()
+        self._register_general_prompts()
+    
+    def _register_analysis_prompts(self) -> None:
+        """Register analysis workflow prompts"""
+        
+        @self.mcp.prompt()
+        def usrn_breakdown_analysis(usrn: str) -> List[PromptMessage]:
+            """Generate a step-by-step USRN breakdown workflow"""
+            template = PROMPT_TEMPLATES["usrn_breakdown"].format(usrn=usrn)
+            
+            return [
+                PromptMessage(
+                    role="user",
+                    content=TextContent(
+                        type="text",
+                        text=f"As an expert in OS NGD API workflows and transport network analysis, {template}"
+                    )
+                )
+            ]
+    
+    def _register_general_prompts(self) -> None:
+        """Register general OS NGD guidance prompts"""
+        
+        @self.mcp.prompt()
+        def collection_query_guidance(collection_id: str, query_type: str = "features") -> List[PromptMessage]:
+            """Generate guidance for querying OS NGD collections"""
+            return [
+                PromptMessage(
+                    role="user",
+                    content=TextContent(
+                        type="text",
+                        text=f"As an OS NGD API expert, guide me through querying the '{collection_id}' collection for {query_type}. "
+                              f"Include: 1) Available filters, 2) Best practices for bbox queries, "
+                              f"3) CRS considerations, 4) Example queries with proper syntax."
+                    )
+                )
+            ]
+        
+        @self.mcp.prompt()
+        def workflow_planning(user_request: str, data_theme: str = "transport") -> List[PromptMessage]:
+            """Generate a workflow plan for complex OS NGD queries"""
+            return [
+                PromptMessage(
+                    role="user",
+                    content=TextContent(
+                        type="text",
+                        text=f"As a geospatial workflow planner, create a detailed workflow plan for: '{user_request}'. "
+                              f"Focus on {data_theme} theme data. Include: "
+                              f"1) Collection selection rationale, "
+                              f"2) Query sequence with dependencies, "
+                              f"3) Filter strategies, "
+                              f"4) Error handling considerations."
+                    )
+                )
+            ] 

src/mcp_service/protocols.py

@@ -9,6 +9,18 @@ def tool(self) -> Callable[..., Any]:
         """Register a function as an MCP tool"""
         ...
 
+    def resource(
+        self,
+        uri: str,
+        *,
+        name: str | None = None,
+        title: str | None = None,
+        description: str | None = None,
+        mime_type: str | None = None,
+    ) -> Callable[[Any], Any]:
+        """Register a function as an MCP resource"""
+        ...
+
     def run(self) -> None:
         """Run the MCP service"""
         ...

src/mcp_service/resources.py

@@ -0,0 +1,80 @@
+"""MCP Resources for OS NGD documentation"""
+
+import json
+import time
+from models import NGDAPIEndpoint
+from utils.logging_config import get_logger
+
+logger = get_logger(__name__)
+
+# TODO: Do this for 
+class OSDocumentationResources:
+    """Handles registration of OS NGD documentation resources"""
+
+    def __init__(self, mcp_service, api_client):
+        self.mcp = mcp_service
+        self.api_client = api_client
+
+    def register_all(self) -> None:
+        """Register all documentation resources"""
+        self._register_transport_network_resources()
+        # Future: self._register_land_resources()
+        # Future: self._register_building_resources()
+
+    def _register_transport_network_resources(self) -> None:
+        """Register transport network documentation resources"""
+
+        @self.mcp.resource("os-docs://street")
+        async def street_docs() -> str:
+            return await self._fetch_doc_resource(
+                "street", NGDAPIEndpoint.MARKDOWN_STREET.value
+            )
+
+        @self.mcp.resource("os-docs://road")
+        async def road_docs() -> str:
+            return await self._fetch_doc_resource(
+                "road", NGDAPIEndpoint.MARKDOWN_ROAD.value
+            )
+
+        @self.mcp.resource("os-docs://tram-on-road")
+        async def tram_on_road_docs() -> str:
+            return await self._fetch_doc_resource(
+                "tram-on-road", NGDAPIEndpoint.TRAM_ON_ROAD.value
+            )
+
+        @self.mcp.resource("os-docs://road-node")
+        async def road_node_docs() -> str:
+            return await self._fetch_doc_resource(
+                "road-node", NGDAPIEndpoint.ROAD_NODE.value
+            )
+
+        @self.mcp.resource("os-docs://road-link")
+        async def road_link_docs() -> str:
+            return await self._fetch_doc_resource(
+                "road-link", NGDAPIEndpoint.ROAD_LINK.value
+            )
+
+        @self.mcp.resource("os-docs://road-junction")
+        async def road_junction_docs() -> str:
+            return await self._fetch_doc_resource(
+                "road-junction", NGDAPIEndpoint.ROAD_JUNCTION.value
+            )
+
+    async def _fetch_doc_resource(self, feature_type: str, url: str) -> str:
+        """Generic method to fetch documentation resources"""
+        try:
+            content = await self.api_client.make_request_no_auth(url)
+
+            return json.dumps(
+                {
+                    "feature_type": feature_type,
+                    "content": content,
+                    "content_type": "markdown",
+                    "source_url": url,
+                    "timestamp": time.time(),
+                }
+            )
+
+        except Exception as e:
+            logger.error(f"Error fetching {feature_type} documentation: {e}")
+            return json.dumps({"error": str(e), "feature_type": feature_type})

src/models.py

@@ -28,6 +28,19 @@ class NGDAPIEndpoint(Enum):
     LINKED_IDENTIFIERS_BASE_PATH = "https://api.os.uk/search/links/v1/{}"
     LINKED_IDENTIFIERS = LINKED_IDENTIFIERS_BASE_PATH.format("identifierTypes/{}/{}")
 
+    # Markdown Resources
+    MARKDOWN_BASE_PATH = "https://docs.os.uk/osngd/data-structure/{}"
+    MARKDOWN_STREET = MARKDOWN_BASE_PATH.format("transport/transport-network/street.md")
+    MARKDOWN_ROAD = MARKDOWN_BASE_PATH.format("transport/transport-network/road.md")
+    TRAM_ON_ROAD = MARKDOWN_BASE_PATH.format(
+        "transport/transport-network/tram-on-road.md"
+    )
+    ROAD_NODE = MARKDOWN_BASE_PATH.format("transport/transport-network/road-node.md")
+    ROAD_LINK = MARKDOWN_BASE_PATH.format("transport/transport-network/road-link.md")
+    ROAD_JUNCTION = MARKDOWN_BASE_PATH.format(
+        "transport/transport-network/road-junction.md"
+    )
+
     # Places API Endpoints
     # PLACES_BASE_PATH = "https://api.os.uk/search/places/v1/{}"
     # PLACES_UPRN = PLACES_BASE_PATH.format("uprn")

src/prompt_templates/prompt_templates.py

@@ -5,7 +5,7 @@
         "Step 1: GET /collections/trn-ntwk-street-1/items?filter=usrn='{usrn}' to get street details. "
         "Step 2: Use Street geometry bbox to query Road Links: GET /collections/trn-ntwk-roadlink-4/items?bbox=[street_bbox] "
         "Step 3: Filter Road Links by Street reference using properties.street_ref matching street feature ID. "
-        "Step 4: For each Road Link: GET /collections/trn-ntwk-roadnode-1/items?filter=roadlink_ref='{roadlink_id}' "
+        "Step 4: For each Road Link: GET /collections/trn-ntwk-roadnode-1/items?filter=roadlink_ref='roadlink_id' "
         "Step 5: Set crs=EPSG:27700 for British National Grid coordinates. "
         "Return: Complete breakdown of USRN into constituent Road Links with node connections."
     ),