create_graph
Create a new knowledge graph with custom configuration including persistence settings, size limits, and descriptive metadata for organizing structured data.
Instructions
Create a new graph with specified configuration.
Args:
graph_id: Unique identifier for the graph
name: Human-readable name for the graph
description: Optional description of the graph's purpose
is_persistent: Whether the graph persists across sessions
max_triples: Optional limit on graph size
ctx: MCP context for user authentication
Returns:
JSON response with graph creation details and next steps
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| graph_id | Yes | ||
| name | Yes | ||
| description | No | ||
| is_persistent | No | ||
| max_triples | No |
Implementation Reference
- Primary handler function for the 'create_graph' MCP tool. Authenticates via context, prepares payload, calls API to create graph, handles response with McpCreateGraphResponse, and returns formatted JSON. Includes comprehensive error handling.@mcp_server.tool() async def create_graph( graph_id: str, name: str, ctx: Context, description: str = "", is_persistent: bool = True, max_triples: int = 0 ) -> str: """ Create a new graph with specified configuration. Args: graph_id: Unique identifier for the graph name: Human-readable name for the graph description: Optional description of the graph's purpose is_persistent: Whether the graph persists across sessions max_triples: Optional limit on graph size ctx: MCP context for user authentication Returns: JSON response with graph creation details and next steps """ operation_id = str(uuid.uuid4()) with logger.operation_context("create_graph", operation_id=operation_id, graph_id=graph_id): try: # Get session context with structured logging auth_token, session_data = await get_session_context(ctx) user_id = auth_token # In development, token is user_id logger.info( "Creating new graph", extra_context={ "user_id": user_id, "graph_id": graph_id, "name": name, "is_persistent": is_persistent } ) # Prepare API request payload payload = { "name": name, "description": description, "is_persistent": is_persistent } if max_triples > 0: payload["max_triples"] = max_triples # Call API server api_response = await api_client._make_request( "POST", f"/graphs/{graph_id}", headers={"Authorization": f"Bearer {auth_token}"}, json=payload ) logger.info( "Graph created successfully", extra_context={ "operation_id": operation_id, "graph_id": graph_id, "user_id": user_id } ) # Transform to enhanced MCP format mcp_response = McpCreateGraphResponse.from_api_response( api_response, operation_id=operation_id, user_id=user_id ) return mcp_response.to_json() except Exception as e: logger.error( "Graph creation failed", extra_context={ "operation_id": operation_id, "graph_id": graph_id, "error": str(e), "error_type": type(e).__name__ } ) error_response = McpErrorResponse.from_exception( e, error_type="GRAPH_CREATION_FAILED", help_info={ "operation_id": operation_id, "graph_id": graph_id, "suggestions": [ "Check that graph_id is unique and valid", "Verify user permissions for graph creation", "Ensure name is not empty" ] } ) return error_response.to_json()
- Response schema class for create_graph tool. Transforms API response into structured MCP format with operation summary, graph details, next steps guidance, and sample SPARQL queries.class McpCreateGraphResponse(McpResponseObject): """Response for graph creation operations""" operation_summary: McpOperationSummary graph_info: Dict[str, Any] next_steps: List[str] sample_queries: List[str] success: bool = True @classmethod def from_api_response(cls, api_response: Dict[str, Any], operation_id: str, user_id: Optional[str] = None) -> 'McpCreateGraphResponse': """Transform API graph creation response to MCP format""" import datetime graph_data = api_response.get("data", {}) graph_id = graph_data.get("graph_id", graph_data.get("id", "")) return cls( operation_summary=McpOperationSummary( operation_type="create_graph", operation_id=operation_id, timestamp=datetime.datetime.now().isoformat(), user_id=user_id ), graph_info={ "id": graph_id, "name": graph_data.get("name", ""), "description": graph_data.get("description", ""), "created_at": graph_data.get("created_at"), "is_persistent": graph_data.get("is_persistent", True), "max_triples": graph_data.get("max_triples") }, next_steps=[ f"Upload RDF data using upload_rdf_file tool", f"Query the graph using sparql_query tool with graph_id='{graph_id}'", f"Explore the schema using get_graph_schema tool" ], sample_queries=[ f"SELECT ?s ?p ?o WHERE {{ ?s ?p ?o }} LIMIT 10", f"SELECT (COUNT(*) as ?count) WHERE {{ ?s ?p ?o }}", f"SELECT DISTINCT ?type WHERE {{ ?s a ?type }}" ] )
- src/neem/mcp/server/standalone_server.py:1079-1081 (registration)Log statement confirming 'create_graph' is registered as an available MCP tool in the standalone server.logger.info("Available tools: create_session, sparql_query, list_graphs, get_graph_schema, create_graph, delete_graph, get_graph_info, get_system_health, upload_file_to_graph") return mcp_server