Create a FastMCP server with graph query tools backed by *store*. When *store* is ``None`` (no database found), every tool returns a friendly "no index" response instead of raising an error.
(store: GraphStore | None)
| 55 | |
| 56 | |
| 57 | def create_mcp_server(store: GraphStore | None) -> FastMCP: |
| 58 | """Create a FastMCP server with graph query tools backed by *store*. |
| 59 | |
| 60 | When *store* is ``None`` (no database found), every tool returns a |
| 61 | friendly "no index" response instead of raising an error. |
| 62 | """ |
| 63 | server = FastMCP("opentrace") |
| 64 | |
| 65 | @server.tool() |
| 66 | def search_graph(query: str, limit: int = 50, nodeTypes: str = "") -> str: |
| 67 | """Full-text search across graph nodes by name or properties. |
| 68 | |
| 69 | Returns matching nodes with their types and properties. |
| 70 | """ |
| 71 | if not store: |
| 72 | logger.info("search_graph called but no index exists") |
| 73 | return NO_INDEX_MSG |
| 74 | logger.debug("search_graph(query=%r, limit=%d, nodeTypes=%r)", query, limit, nodeTypes) |
| 75 | try: |
| 76 | node_types = [t.strip() for t in nodeTypes.split(",") if t.strip()] or None |
| 77 | limit = min(limit, 1000) |
| 78 | nodes = store.search_nodes(query, node_types=node_types, limit=limit) |
| 79 | logger.debug("search_graph → %d results", len(nodes)) |
| 80 | return _json_response(nodes) |
| 81 | except Exception as e: |
| 82 | return _error_response("search_graph", e) |
| 83 | |
| 84 | @server.tool() |
| 85 | def list_nodes(type: str, limit: int = 50, filters: dict[str, Any] | None = None) -> str: |
| 86 | """List nodes of a specific type. |
| 87 | |
| 88 | Valid types include: Repository, Class, Function, File, Directory, |
| 89 | Package, Module, Service, Endpoint, Database. |
| 90 | """ |
| 91 | if not store: |
| 92 | logger.info("list_nodes called but no index exists") |
| 93 | return NO_INDEX_MSG |
| 94 | logger.debug("list_nodes(type=%r, limit=%d, filters=%r)", type, limit, filters) |
| 95 | try: |
| 96 | limit = min(limit, 1000) |
| 97 | nodes = store.list_nodes(node_type=type, filters=filters, limit=limit) |
| 98 | logger.debug("list_nodes → %d results", len(nodes)) |
| 99 | return _json_response(nodes) |
| 100 | except Exception as e: |
| 101 | return _error_response("list_nodes", e) |
| 102 | |
| 103 | @server.tool() |
| 104 | def get_node(nodeId: str) -> str: |
| 105 | """Get full details of a single node by its ID, including all properties and immediate neighbors.""" |
| 106 | if not store: |
| 107 | logger.info("get_node called but no index exists") |
| 108 | return NO_INDEX_MSG |
| 109 | logger.debug("get_node(nodeId=%r)", nodeId) |
| 110 | try: |
| 111 | node = store.get_node(nodeId) |
| 112 | if node is None: |
| 113 | return json.dumps({"error": f"Node not found: {nodeId}"}) |
| 114 |
no outgoing calls