feature/allow index generation from resource docs

Author: nemozak1

database/obp_utils.py

@@ -6,26 +6,47 @@
 from typing import Dict, Any
 
 
-def get_obp_config(endpoint_type="static"):
-    """Get OBP configuration from environment variables."""
+def get_obp_config(endpoint_type: str = "static", use_resource_docs: bool = True) -> Dict[str, str]:
+    """
+    Get OBP configuration from environment variables.
+    
+    Args:
+        endpoint_type: Type of endpoints to fetch ("static", "dynamic", or "all")
+        use_resource_docs: If True, use resource-docs endpoint (includes roles).
+                          If False, use swagger endpoint.
+    
+    Returns:
+        Dictionary with configuration including URLs for data fetching
+    """
     base_url = os.getenv("OBP_BASE_URL")
     api_version = os.getenv("OBP_API_VERSION")
 
     if not all([base_url, api_version]):
         raise ValueError("Missing required environment variables: OBP_BASE_URL, OBP_API_VERSION")
 
-    return {
+    config = {
         "base_url": base_url,
         "api_version": api_version,
         "glossary_url": f"{base_url}/obp/{api_version}/api/glossary",
-        "swagger_url": f"{base_url}/obp/{api_version}/resource-docs/{api_version}/swagger?content={endpoint_type}"
+        "swagger_url": f"{base_url}/obp/{api_version}/resource-docs/{api_version}/swagger?content={endpoint_type}",
+        "resource_docs_url": f"{base_url}/obp/{api_version}/resource-docs/{api_version}/obp?content={endpoint_type}",
     }
+    
+    # Set the primary data URL based on preference
+    if use_resource_docs:
+        config["data_url"] = config["resource_docs_url"]
+        config["data_format"] = "resource_docs"
+    else:
+        config["data_url"] = config["swagger_url"]
+        config["data_format"] = "swagger"
+    
+    return config
 
 
 def fetch_obp_data(url: str) -> Dict[str, Any]:
     """Fetch data from OBP API endpoint."""
     print(f"Fetching data from: {url}")
-    response = requests.get(url, timeout=30)
+    response = requests.get(url, timeout=60)  # Increased timeout for large responses
     try:
         response.raise_for_status()
     except requests.exceptions.HTTPError as e:

scripts/generate_endpoint_index.py

@@ -1,9 +1,10 @@
 #!/usr/bin/env python3
 """
-Script to generate the lightweight endpoint index from OBP swagger data.
+Script to generate the lightweight endpoint index from OBP resource docs.
 
-This script fetches the swagger/OpenAPI specification from the OBP API
-and generates a lightweight index for fast endpoint discovery.
+This script fetches the resource docs from the OBP API and generates a 
+lightweight index for fast endpoint discovery. Resource docs provide richer
+information than swagger, including roles/entitlements required for each endpoint.
 """
 import os
 import sys
@@ -23,7 +24,7 @@ def main():
 
     # Parse command line arguments
     parser = argparse.ArgumentParser(
-        description="Generate lightweight endpoint index from OBP swagger data."
+        description="Generate lightweight endpoint index from OBP resource docs."
     )
     parser.add_argument(
         "--endpoints",
@@ -37,33 +38,48 @@ def main():
         default=None,
         help="Output file path for the index (default: database/endpoint_index.json)"
     )
+    parser.add_argument(
+        "--use-swagger",
+        action="store_true",
+        help="Use swagger endpoint instead of resource docs (not recommended - lacks role info)"
+    )
     args = parser.parse_args()
 
+    use_resource_docs = not args.use_swagger
+    
     try:
         # Get configuration
-        config = get_obp_config(args.endpoints)
+        config = get_obp_config(args.endpoints, use_resource_docs=use_resource_docs)
 
-        # Fetch swagger data
+        # Fetch data
         print("\n" + "="*50)
-        print("FETCHING SWAGGER DATA")
+        print(f"FETCHING {'RESOURCE DOCS' if use_resource_docs else 'SWAGGER'} DATA")
         print("="*50)
-        swagger_data = fetch_obp_data(config["swagger_url"])
+        print(f"URL: {config['data_url']}")
+        
+        data = fetch_obp_data(config["data_url"])
 
         # Build index
         print("\n" + "="*50)
         print("BUILDING ENDPOINT INDEX")
         print("="*50)
 
         index = EndpointIndex(index_file=args.output)
-        index.build_index_from_swagger(swagger_data)
+        
+        if use_resource_docs:
+            index.build_index_from_resource_docs(data)
+        else:
+            index.build_index_from_swagger(data)
 
         # Print statistics
         print("\n" + "="*50)
         print("INDEX GENERATION COMPLETE")
         print("="*50)
+        print(f"Data source: {'Resource Docs' if use_resource_docs else 'Swagger'}")
         print(f"Total endpoints indexed: {len(index._index)}")
         print(f"Total unique tags: {len(index.get_all_tags())}")
-        print(f"Output file: {index.index_file}")
+        print(f"Index file: {index.index_file}")
+        print(f"Schemas file: {index.schemas_file}")
 
         # Print sample of tags
         all_tags = index.get_all_tags()
@@ -72,8 +88,18 @@ def main():
             for tag in all_tags[:10]:
                 print(f"  - {tag}")
 
+        # If using resource docs, show role statistics
+        if use_resource_docs:
+            endpoints_with_roles = sum(
+                1 for schema in index._schemas.values() 
+                if schema.get("roles")
+            )
+            print(f"\nEndpoints with role requirements: {endpoints_with_roles}")
+        
     except Exception as e:
         print(f"Error during index generation: {e}")
+        import traceback
+        traceback.print_exc()
         sys.exit(1)
 
 

src/mcp_server_obp/elicitation.py

@@ -9,9 +9,14 @@ class ApprovalConsent:
     consent_id: str
 
 
-def get_roles(endpoint: dict[str, Any]) -> list[str]:
+def get_roles_for_endpoint(endpoint_id: str) -> list[str]:
     """
-    Gets the required entitlements for a given OBP endpoint.
+    Gets the required entitlements for a given OBP endpoint from its ID.
     """
 
-    
+    index = get_endpoint_index()
+    endpoint = index.get_endpoint_schema(endpoint_id)
+    if not endpoint:
+        raise ValueError(f"Endpoint with ID '{endpoint_id}' not found in index.")
+    
+    return endpoint.roles

src/tools/__init__.py

@@ -6,6 +6,7 @@
     EndpointSchema,
     EndpointParameter,
     EndpointResponse,
+    RoleRequirement,
     HttpMethod,
     get_endpoint_index,
     reload_endpoint_index,
@@ -17,6 +18,7 @@
     "EndpointSchema",
     "EndpointParameter",
     "EndpointResponse",
+    "RoleRequirement",
     "HttpMethod",
     "get_endpoint_index",
     "reload_endpoint_index",

src/tools/endpoint_index.py

@@ -105,24 +105,45 @@ class SecurityRequirement(BaseModel):
     }
 
 
+class RoleRequirement(BaseModel):
+    """
+    OBP Role requirement for an endpoint.
+    
+    Represents a required role/entitlement to access the endpoint.
+    """
+    role: str = Field(..., description="The role/entitlement name")
+    requires_bank_id: bool = Field(default=False, description="Whether this role requires a bank_id context")
+    
+    model_config = {
+        "extra": "ignore",
+    }
+
+
 class EndpointSchema(BaseModel):
     """
-    Full OpenAPI schema for an endpoint.
+    Full schema for an endpoint.
     
     Contains complete specification including parameters, request body,
-    responses, and security requirements.
+    responses, security requirements, and roles from OBP resource docs.
     """
     path: str = Field(..., description="API path template")
     method: HttpMethod = Field(..., description="HTTP method")
     operation_id: str = Field(default="", description="OpenAPI operationId")
     summary: str = Field(default="", description="Brief description")
     description: str = Field(default="", description="Detailed description (may contain HTML)")
+    description_markdown: str = Field(default="", description="Description in markdown format")
     tags: List[str] = Field(default_factory=list, description="Tags for categorization")
     parameters: List[EndpointParameter] = Field(default_factory=list, description="Endpoint parameters")
     requestBody: Optional[Dict[str, Any]] = Field(default=None, description="Request body specification")
     responses: Dict[str, EndpointResponse] = Field(default_factory=dict, description="Response specifications by status code")
+    success_response_body: Optional[Dict[str, Any]] = Field(default=None, description="Example success response body")
+    error_response_bodies: List[str] = Field(default_factory=list, description="Possible error response messages")
+    typed_success_response_body: Optional[Dict[str, Any]] = Field(default=None, description="JSON Schema for success response")
     security: List[Dict[str, Any]] = Field(default_factory=list, description="Security requirements")
-    roles: List[str] = Field(default_factory=list, description="Required roles")
+    roles: List[RoleRequirement] = Field(default_factory=list, description="Required roles/entitlements")
+    is_featured: bool = Field(default=False, description="Whether this endpoint is featured")
+    special_instructions: str = Field(default="", description="Special instructions for using this endpoint")
+    connector_methods: List[str] = Field(default_factory=list, description="Related connector methods")
 
     model_config = {
         "extra": "ignore",
@@ -150,18 +171,33 @@ def from_raw(cls, data: Dict[str, Any]) -> "EndpointSchema":
             if isinstance(param_data, dict):
                 parameters.append(EndpointParameter(**param_data))
 
+        # Convert roles to RoleRequirement models
+        roles = []
+        for role_data in data.get("roles", []):
+            if isinstance(role_data, dict):
+                roles.append(RoleRequirement(**role_data))
+            elif isinstance(role_data, str):
+                roles.append(RoleRequirement(role=role_data))
+        
         return cls(
             path=data.get("path", ""),
             method=HttpMethod(data.get("method", "GET").upper()),
             operation_id=data.get("operation_id", ""),
             summary=data.get("summary", ""),
             description=data.get("description", ""),
+            description_markdown=data.get("description_markdown", ""),
             tags=data.get("tags", []),
             parameters=parameters,
             requestBody=data.get("requestBody"),
             responses=responses,
+            success_response_body=data.get("success_response_body"),
+            error_response_bodies=data.get("error_response_bodies", []),
+            typed_success_response_body=data.get("typed_success_response_body"),
             security=data.get("security", []),
-            roles=data.get("roles", []),
+            roles=roles,
+            is_featured=data.get("is_featured", False),
+            special_instructions=data.get("special_instructions", ""),
+            connector_methods=data.get("connector_methods", []),
         )
 
 
@@ -330,6 +366,88 @@ def build_index_from_swagger(self, swagger_data: Dict[str, Any]) -> None:
         self._save_schemas()
         self._schemas_loaded = True
 
+    def build_index_from_resource_docs(self, resource_docs_data: Dict[str, Any]) -> None:
+        """
+        Build the lightweight endpoint index and full schemas from OBP resource docs.
+        
+        Resource docs provide richer information than swagger, including:
+        - Roles/entitlements required for each endpoint
+        - Markdown descriptions
+        - Example success/error response bodies
+        - Typed response schemas
+        - Connector methods
+        
+        Args:
+            resource_docs_data: Resource docs response from OBP API
+        """
+        self._index = {}
+        self._schemas = {}
+        self._index_models = {}
+        self._schema_models = {}
+        
+        resource_docs = resource_docs_data.get("resource_docs", [])
+        
+        logger.info(f"Building index from {len(resource_docs)} resource docs...")
+        
+        for doc in resource_docs:
+            if not isinstance(doc, dict):
+                continue
+            
+            # Extract fields from resource doc format
+            operation_id = doc.get("operation_id", "")
+            method = doc.get("request_verb", "GET").upper()
+            path = doc.get("request_url", "")
+            summary = doc.get("summary", "")
+            tags = doc.get("tags", [])
+            
+            # Use operation_id as the endpoint ID (fallback to generated ID if not present)
+            if operation_id:
+                endpoint_id = operation_id
+            else:
+                endpoint_id = self._generate_endpoint_id(method, path)
+                logger.warning(f"No operation_id for {method} {path}, using generated ID: {endpoint_id}")
+            
+            # Store lightweight index entry
+            self._index[endpoint_id] = {
+                "id": endpoint_id,
+                "method": method,
+                "path": path,
+                "operation_id": operation_id,
+                "summary": summary,
+                "tags": tags
+            }
+            
+            # Extract roles from resource docs format
+            roles = doc.get("roles", [])
+            
+            # Store full schema separately (keyed by endpoint_id)
+            self._schemas[endpoint_id] = {
+                "path": path,
+                "method": method,
+                "operation_id": operation_id,
+                "summary": summary,
+                "description": doc.get("description", ""),
+                "description_markdown": doc.get("description_markdown", ""),
+                "tags": tags,
+                "parameters": [],  # Resource docs don't provide parameters in the same way
+                "requestBody": None,
+                "responses": {},
+                "success_response_body": doc.get("success_response_body"),
+                "error_response_bodies": doc.get("error_response_bodies", []),
+                "typed_success_response_body": doc.get("typed_success_response_body"),
+                "security": [],
+                "roles": roles,
+                "is_featured": doc.get("is_featured", False),
+                "special_instructions": doc.get("special_instructions", ""),
+                "connector_methods": doc.get("connector_methods", []),
+            }
+        
+        logger.info(f"Built index with {len(self._index)} endpoints")
+        logger.info(f"Built schemas for {len(self._schemas)} endpoints")
+        self._save_index()
+        self._save_schemas()
+        self._schemas_loaded = True
+    
     def _generate_endpoint_id(self, method: str, path: str) -> str:
         """Generate a unique ID for an endpoint."""
         # Clean the path for use in ID