get_graph_schema
Retrieve schema information for a specific knowledge graph to understand its structure and data organization.
Instructions
Get schema information for a specific graph via API server.
Args:
graph_id: Graph ID to analyze
Returns:
JSON string with schema information
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| graph_id | Yes |
Implementation Reference
- MCP tool handler for get_graph_schema: authenticates via context, validates graph access using session data, calls API client to fetch schema, transforms to McpSchemaResponse, returns formatted JSON or error.@mcp_server.tool() async def get_graph_schema( graph_id: str, ctx: Context ) -> str: """ Get schema information for a specific graph via API server. Args: graph_id: Graph ID to analyze Returns: JSON string with schema information """ try: # Get session context (cached or fresh) auth_token, session_data = await get_session_context(ctx) # Validate graph access if we have session data if session_data: accessible_graphs = session_data.get("accessible_graphs", []) if graph_id not in accessible_graphs: error_response = McpErrorResponse( error_type="GRAPH_ACCESS_DENIED", error_message=f"Graph '{graph_id}' not accessible to user", help={ "accessible_graphs": accessible_graphs, "suggestions": ["Use list_graphs to see available graphs"] } ) return error_response.to_json() logger.debug(f"Graph '{graph_id}' validated via session cache") logger.info(f"Getting schema for graph '{graph_id}' via API") # Call API server api_response = await api_client.get_graph_schema(graph_id, auth_token) # Transform to enhanced MCP format mcp_response = McpSchemaResponse.from_api_response(api_response, graph_id) return mcp_response.to_json() except Exception as e: logger.error(f"Schema analysis failed: {e}") error_response = McpErrorResponse.from_exception( e, error_type="SCHEMA_ANALYSIS_FAILED", help_info={ "graph_id": graph_id, "suggestions": [ "Check graph exists with list_graphs tool", "Verify graph access permissions" ] } ) return error_response.to_json()
- Output schema definition as McpSchemaResponse dataclass, including summary stats, enhanced classes with sample queries, and standard query templates. Includes factory method to transform raw API schema data.class McpSchemaResponse(McpResponseObject): """Enhanced schema response""" graph_id: str schema_summary: McpSchemaSummary classes: List[McpSchemaClass] query_templates: Dict[str, str] success: bool = True @classmethod def from_api_response(cls, api_response: Dict[str, Any], graph_id: str) -> 'McpSchemaResponse': """Transform API schema response to enhanced MCP format""" schema_data = api_response.get("data", {}) # Transform classes with simple enhancements classes = [] for class_data in schema_data.get("classes", []): classes.append(McpSchemaClass( uri=class_data.get("uri", ""), label=class_data.get("label", "") or _extract_local_name(class_data.get("uri", "")), description=class_data.get("description"), instance_count=class_data.get("instance_count"), sample_query=_generate_sample_query(class_data.get("uri", "")) )) return cls( graph_id=graph_id, schema_summary=McpSchemaSummary( total_classes=len(classes), total_properties=len(schema_data.get("properties", [])), primary_namespaces=_extract_namespaces(schema_data) ), classes=classes, query_templates={ "list_instances": "SELECT ?instance ?label WHERE { ?instance a <CLASS_URI> . OPTIONAL { ?instance rdfs:label ?label } } LIMIT 20", "explore_properties": "SELECT ?property ?value WHERE { <INSTANCE_URI> ?property ?value }", "count_by_type": "SELECT ?type (COUNT(?instance) as ?count) WHERE { ?instance a ?type } GROUP BY ?type ORDER BY DESC(?count)" } )
- Helper method in MCPAPIClient that performs the HTTP GET request to the backend API's schema endpoint /graphs/{graph_id}/schema using the user's auth token.async def get_graph_schema(self, graph_id: str, auth_token: str) -> Dict[str, Any]: """Call API /graphs/{graph_id}/schema endpoint.""" headers = {"Authorization": f"Bearer {auth_token}"} endpoint = f"/graphs/{graph_id}/schema" return await self._make_request("GET", endpoint, headers)
- src/neem/mcp/server/standalone_server.py:474-474 (registration)Tool registration decorator @mcp_server.tool() that registers get_graph_schema as an MCP tool with FastMCP server, inferring input schema from function signature.@mcp_server.tool()