diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..9c43622 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,19 @@ +**/__pycache__/ +**/.git +**/.github +**/.mypy_cache +**/.pytest_cache +**/.vscode +**/.idea +**/.coverage +**/.DS_Store +**/.python-version +**/*.bak +**/*.pyc +**/node_modules/ + +tests/ +docs/ +install/ +venv +.venv diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md new file mode 120000 index 0000000..f811a97 --- /dev/null +++ b/.github/copilot-instructions.md @@ -0,0 +1 @@ +LLMS.md \ No newline at end of file diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..abd52bf --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,14 @@ +--- +repos: + - repo: https://github.com/pre-commit/mirrors-mypy + rev: v1.15.0 + hooks: + - id: mypy + name: mypy + args: [--show-error-codes src/] + - repo: https://github.com/astral-sh/ruff-pre-commit + rev: v0.12.4 + hooks: + - id: ruff-format + - id: ruff + args: [--fix] diff --git a/CLAUDE.md b/CLAUDE.md new file mode 120000 index 0000000..f811a97 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1 @@ +LLMS.md \ No newline at end of file diff --git a/GEMINI.md b/GEMINI.md new file mode 120000 index 0000000..f811a97 --- /dev/null +++ b/GEMINI.md @@ -0,0 +1 @@ +LLMS.md \ No newline at end of file diff --git a/GPT.md b/GPT.md new file mode 120000 index 0000000..f811a97 --- /dev/null +++ b/GPT.md @@ -0,0 +1 @@ +LLMS.md \ No newline at end of file diff --git a/LLMS.md b/LLMS.md new file mode 100644 index 0000000..ba51c3e --- /dev/null +++ b/LLMS.md @@ -0,0 +1,49 @@ +# LLM Context Guide for Infrahub MCP Server + +Infrahub MCP Server connects your AI assistants to Infrahub using the open MCP standard—so agents can read and (optionally) change your infra state through a consistent, audited, human-approved interface. + +## Key Directories + +``` +infrahub-mcp/ +├── docs/ # Documentation (UPDATE FOR CHANGES) +├── src/ # Python Code +│   └── infrahub_mcp/ +│ ├── __init__.py +| ├── branch.py +| ├── constants.py +| ├── gql.py +| ├── nodes.py +| ├── prompts/ +| │   └── main.md +| ├── schema.py +| ├── server.py +| └── utils.py +└── tests/ # Python/integration tests +``` + +## Code Standards + +### Python Backend + +- **Type hints required** for all new code +- **MyPy compliant** - run `pre-commit run mypy` +- **Ruf compliant**- run `pre-commit run ruff` +- **pytest** for testing + +## Documentation Requirements + +- **docs/**: Update for any user-facing changes +- **Docstrings**: Required for new functions/classes + +## Test Utilities + +### Running Tests + +## Platform-Specific Instructions + +- **[CLAUDE.md](CLAUDE.md)** - For Claude/Anthropic tools +- **[.github/copilot-instructions.md](.github/copilot-instructions.md)** - For GitHub Copilot +- **[GEMINI.md](GEMINI.md)** - For Google Gemini tools +- **[GPT.md](GPT.md)** - For OpenAI/ChatGPT tools +- **[.cursor/rules/dev-standard.mdc](.cursor/rules/dev-standard.mdc)** - For Cursor editor diff --git a/README.md b/README.md index fae42d9..f057c75 100644 --- a/README.md +++ b/README.md @@ -1,150 +1,19 @@ -# Infrahub MCP Server + +![Infrahub Logo](https://assets-global.website-files.com/657aff4a26dd8afbab24944b/657b0e0678f7fd35ce130776_Logo%20INFRAHUB.svg) + -MCP server to interact with Infrahub +# Infrahub MCP -## Requirements +[Infrahub](https://github.com/opsmill/infrahub) by [OpsMill](https://opsmill.com) acts as a central hub to manage the data, templates and playbooks that powers your infrastructure. At its heart, Infrahub is built on 3 fundamental pillars: -- Python 3.13+ -- fastmcp -- infrahub_sdk +- **A Flexible Schema**: A model of the infrastructure and the relation between the objects in the model, that's easily extensible. +- **Version Control**: Natively integrated into the graph database which opens up some new capabilities like branching, diffing, and merging data directly in the database. +- **Unified Storage**: By combining a graph database and git, Infrahub stores data and code needed to manage the infrastructure. +## Introduction -## Installation +Infrahub MCP Server connects your AI assistants to Infrahub using the open MCP standard—so agents can read and (optionally) change your infra state through a consistent, audited, human-approved interface. -1. **Clone the repo** - - ```bash - git clone https://github.com/opsmill/infrahub-mcp-server.git - cd infrahub-mcp-server - ``` - -2. **Install dependencies** - - ```bash - uv sync - ``` - -3. **Run the server** - - ```bash - uv run fastmcp run src/infrahub_mcp_server/server.py:mcp - ``` - - -## Configuration - -Set the following environment variables as needed: - -| Variable | Description | Default | -|---------------------|-------------------------------------|--------------------------| -| `INFRAHUB_ADDRESS` | URL of your Infrahub instance | `http://localhost:8000` | -| `INFRAHUB_API_TOKEN`| API token for Infrahub | `placeholder UUID` | -| `MCP_HOST` | Host for the web server | `0.0.0.0` | -| `MCP_PORT` | Port for the web server | `8001` | - - -## Usage - -### HTTP API mode - -```bash -poetry run python server.py --web -``` - -Send a POST request to the root endpoint (/): - -```bash -curl -X POST http://localhost:8001/ \\ - -H "Content-Type: application/json" \\ - -d '{ - "tool": "infrahub_get_nodes", - "params": { - "kind": "Tag", - "filters": { "any": "blue" }, - "partial_match": false - } - }' -``` - -### CLI / stdin-stdout mode - -```bash -uv run fastmcp run src/infrahub_mcp_server/server.py:mcp -``` - -Then send JSON-RPC requests to stdin: - -```bash -{"jsonrpc":"2.0","method":"tools/discover","params":{}} -{"jsonrpc":"2.0","method":"tools/call","params":{"name": "infrahub_get_nodes","arguments": {"kind":"Router","filters": {"location":"eu-west"}}}} -``` - -Process one-shot request with --oneshot: - -```bash -echo '{"method":"tools/discover"}' | poetry run python server.py --oneshot -``` - -## MCP Methods - -### infrahub_get_nodes -Retrieve nodes of a given kind. - -Params: - -- kind (string, required): node kind (e.g. "Router") -- branch (string): branch name -- filters (object): key/value filters -- simple keys → __value -- list values → __values -- use "any" to search across all attributes -- partial_match (boolean): true for substring matches -- infrahub_url (string): override via param -- infrahub_api_token (string): override via param - -Response: - -```json -{ - "success": true, - "count": 5, - "nodes": [ { ... }, ... ] -} -``` - -### infrahub_get_schema - -Retrieve schema details. - -Params: - -- kind (string): kind name; omit to fetch all schemas -- branch (string) -- exclude_profiles (boolean) -- exclude_templates (boolean) -- infrahub_url (string) -- infrahub_api_token (string) - -Response (single kind): - -```json -{ - "success": true, - "kind": "Tag", - "namespace": "...", - "name": "...", - "attributes": [ { "name": "...", "type": "...", "description": "..." }, ... ], - "relationships": [ { "name": "...", "rel_kind": "...", "cardinality": "...", "description": "..." }, ... ] -} -``` - -Response (all schemas): - -```json -{ - "success": true, - "count": 12, - "schemas": [ { ... }, ... ] -} -``` +## Using Infrahub MCP +Documentation for using Infrahub Sync is available [here](https://docs.infrahub.app/mcp/) diff --git a/docs/docs/guides/installation.mdx b/docs/docs/guides/installation.mdx new file mode 100644 index 0000000..697fcf2 --- /dev/null +++ b/docs/docs/guides/installation.mdx @@ -0,0 +1,148 @@ +--- +title: Installing Infrahub MCP +--- + +import Tabs from '@theme/Tabs'; +import TabItem from '@theme/TabItem'; + +This guide provides step-by-step instructions for installing and configuring different MCP clients to connect to the Infrahub MCP server. + +## Prerequisites + +- Ensure you have access to an Infrahub MCP server endpoint +- Each client may have additional prerequisites listed in their respective sections + +## Install & run the Infrahub MCP server + +1. **Clone the repository** + + ```bash + git clone https://github.com/opsmill/infrahub-mcp-server.git + cd infrahub-mcp-server + ``` + +2. **Install dependencies** + + +- Python 3.13+ +- fastmcp +- infrahub_sdk + + + ```bash + uv sync + ``` + +3. **Run the server** + + ```bash + uv run fastmcp run src/infrahub_mcp_server/server.py:mcp + ``` + + +### Configuration + +Set the following environment variables as needed: + +| Variable | Description | Default | +|---------------------|-------------------------------------|--------------------------| +| `INFRAHUB_ADDRESS` | URL of your Infrahub instance | `http://localhost:8000` | +| `INFRAHUB_API_TOKEN`| API token for Infrahub | `placeholder UUID` | +| `MCP_HOST` | Host for the web server | `0.0.0.0` | +| `MCP_PORT` | Port for the web server | `8001` | + +## Add to your MCP client + + + + +### Manual steps: + +- Go to **Settings > Cursor Settings > Tools & Integrations** +- Under **MCP tools**, click **Add Custom MCP** +- Paste the configuration below into mcp.json +- **MCP tools, click Add Custom MCP** +- Save the file to apply the configuration +- Restart Cursor if prompted + +### Configuration: + +```json +{ + "mcpServers": { + "infrahub_mcp": { + "command": "uv", + "args": [ + "run", + "fastmcp", + "run", + "src/infrahub_mcp/server.py:mcp" + ], + "env": { + "PYENV_VERSION": "3.13.3", + "INFRAHUB_ADDRESS": "http://localhost:8000", + "INFRAHUB_API_TOKEN": "06438eb2-8019-4776-878c-0941b1f1d1ec", + } + } + } +} +``` + + + + +### Manual steps: + +Use the command line: + +```console +code --add-mcp '{"type":"stdio","name":"infrahub-mcp","version":"0.0.1","description":"MCP server to interact with Infrahub","command":"uv","args":["run","fastmcp","run","src/infrahub_mcp/server.py:mcp"],"author":"Opsmill","tags":["infrahub-mcp","mcp","server"],"categories":["mcp"],"env":{"MCP_HOST":"0.0.0.0","MCP_PORT":"8001","INFRAHUB_ADDRESS":"http://localhost:8000","INFRAHUB_API_TOKEN":"placeholder UUID"}}' +``` + +Then open the MCP Servers configuration file (Cmd+Shift+P and type "MCP: Open User Configuration") and replace the default env values with your own. + +Then go to **Extensions**, find the **Infrahub MCP** server in the list, open the menu with right-click and click on **Start Server**. + + + + +### Manual steps: + +- Go to **Settings > Developer** + +- Click **Edit config** to open the **claude_desktop_config.json** file + +- Add the MCP server configuration to the `mcpServers` section +- Paste the configuration below +- **Modify the default env values with your own** +- Save the file to apply the configuration +- Restart Claude Desktop + +### Configuration: + +```json +{ + "mcpServers": { + "infrahub_mcp": { + "transport": "stdio", + "command": "uv", + "args": [ + "run", + "fastmcp", + "run", + "src/infrahub_mcp/server.py:mcp" + ], + "env": { + "PYENV_VERSION": "3.13.3", + "INFRAHUB_ADDRESS": "http://localhost:8000", + "INFRAHUB_API_TOKEN": "06438eb2-8019-4776-878c-0941b1f1d1ec", + } + } + } +} +``` + + + + +## Related resources diff --git a/docs/docs/readme.mdx b/docs/docs/readme.mdx index 6e34c99..5bca575 100644 --- a/docs/docs/readme.mdx +++ b/docs/docs/readme.mdx @@ -2,4 +2,13 @@ title: Infrahub MCP Server --- -TODO \ No newline at end of file +Infrahub MCP Server lets AI assistants and IDE agents talk to your Infrahub “source-of-truth” safely and in a standardized way. +It implements the Model Context Protocol (MCP) so any MCP-compatible client (agent mode in IDEs, chat assistants, automation runners, etc.) can discover tools, request context, and perform actions against your Infrahub instance—without custom glue code. + +## Guides + +- [Install and run Infrahub MCP](./guides/installation.mdx) + +## Reference + +- [Available methods](./references/methods.mdx) \ No newline at end of file diff --git a/docs/docs/references/methods.mdx b/docs/docs/references/methods.mdx new file mode 100644 index 0000000..5345878 --- /dev/null +++ b/docs/docs/references/methods.mdx @@ -0,0 +1,183 @@ +--- +title: Infrahub MCP methods +--- + +## Schema + + +### get_schema_mapping + + +**Capabilities** +- read-only +- idempotent +- no destroy + +List all schema nodes and generics available in Infrahub. + +**Parameters** + +- **branch** (string): branch to read from; default is the server’s default branch + + +### get_schema + + +**Capabilities** +- read-only +- idempotent +- no destroy + +Retrieve the full schema for a specific kind (attributes, relationships, and types). + +**Parameters** + +- **kind** (string, required): schema Kind to retrieve +- **branch** (string): branch to read from; default is the server’s default branch + + +### get_schemas + + +**Capabilities** +- read-only +- idempotent +- no destroy + +Retrieve all schemas, optionally excluding Profiles and Templates. + +**Parameters** + + +- **branch** (string): branch to read from; default is the server’s default branch +- **exclude_profiles** (boolean, default: true): omit Profile schemas +- **exclude_templates** (boolean, default: true): omit Template schemas + + +## Nodes + + +### get_nodes + + +**Capabilities** +- read-only +- idempotent +- no destroy + +Get all objects of a specific kind from Infrahub. +Use `get_schema_mapping` to discover kinds and `get_node_filters` to discover filter keys. + +**Parameters** + + +- **kind** (string, required): kind of the objects to retrieve +- **branch** (string): branch name (default: server’s default branch) +- **filters** (object): dictionary of filters to apply +- **partial_match** (boolean, default: false): enable substring matches for string filters + + +**Response** + + +### get_node_filters + + +**Capabilities** +- read-only +- idempotent +- no destroy + +Retrieve all available filters for a specific schema node kind. + +**Parameters** + +- **kind** (string, required): node kind (example: "Router") +- **branch** (string): branch name + +**Notes** + +- Attribute filters: `attribute__value` +- Relationship filters: `relationship__attribute__value` +- Filters starting with `parent__` refer to a related generic node +- Use `get_schema` to inspect peer kinds and attributes used in relationship filters + + +### get_related_nodes + + +**Capabilities** +- read-only +- idempotent +- no destroy + +Retrieve related nodes by relation name for a given kind. + +**Parameters** + +- **kind** (string, required): source node kind (example: "Device") +- **relation** (string, required): relationship name to follow (example: "interfaces") +- **filters** (object): filters applied to the **source** nodes before following the relation +- **branch** (string): branch name + +## Branches + + +### branch_create + + +**Capabilities** +- write +- idempotent +- no destroy + +Create a new branch in Infrahub. + +**Parameters** + + +- **name** (string, required): name of the branch to create +- **sync_with_git** (boolean, default: false): whether to sync the branch with Git + + + +### get_branches + + +**Capabilities** +- read-only +- idempotent +- no destroy + +Retrieve all branches from Infrahub. + +## GraphQL + + +### get_graphql_schema + + +**Capabilities** +- read-only +- idempotent +- no destroy + +Retrieve the GraphQL schema from Infrahub as a string. + +**Parameters** +_none_ + + +### query_graphql + + +**Capabilities** +- may write (depends on operation) +- idempotent for queries; non-idempotent for mutations +- destructive potential depends on mutation + +Execute a GraphQL operation against Infrahub. + +**Parameters** + +- **query** (string, required): GraphQL document (use `query { ... }` for reads, `mutation { ... }` for writes) diff --git a/docs/sidebars.ts b/docs/sidebars.ts index fa377b6..6056455 100644 --- a/docs/sidebars.ts +++ b/docs/sidebars.ts @@ -3,6 +3,8 @@ import type {SidebarsConfig} from '@docusaurus/plugin-content-docs'; const sidebars: SidebarsConfig = { mcpSidebar: [ 'readme', + 'guides/installation', + 'references/methods', ] }; diff --git a/src/infrahub_mcp_server/__init__.py b/src/infrahub_mcp_server/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/infrahub_mcp_server/branch.py b/src/infrahub_mcp_server/branch.py deleted file mode 100644 index 8975433..0000000 --- a/src/infrahub_mcp_server/branch.py +++ /dev/null @@ -1,63 +0,0 @@ -from typing import TYPE_CHECKING, Annotated - -from fastmcp import Context, FastMCP -from infrahub_sdk.branch import BranchData -from infrahub_sdk.exceptions import GraphQLError -from mcp.types import ToolAnnotations -from pydantic import Field - -from infrahub_mcp_server.utils import MCPResponse, MCPToolStatus, _log_and_return_error - -if TYPE_CHECKING: - from infrahub_sdk import InfrahubClient - -mcp: FastMCP = FastMCP(name="Infrahub Branches") - - -@mcp.tool( - tags=["branches", "create"], - annotations=ToolAnnotations(readOnlyHint=False, idempotentHint=True, destructiveHint=False), -) -async def branch_create( - ctx: Context, - name: Annotated[str, Field(description="Name of the branch to create.")], - sync_with_git: Annotated[bool, Field(default=False, description="Whether to sync the branch with git.")], -) -> MCPResponse[dict[str, str]]: - """Create a new branch in infrahub. - - Parameters: - name: Name of the branch to create. - sync_with_git: Whether to sync the branch with git. Defaults to False. - - Returns: - Dictionary with success status and branch details. - """ - - client: InfrahubClient = ctx.request_context.lifespan_context.client - ctx.info(f"Creating branch {name} in Infrahub...") - - try: - branch = await client.branch.create(branch_name=name, sync_with_git=sync_with_git, background_execution=False) - - except GraphQLError as exc: - return _log_and_return_error(ctx=ctx, error=exc, remediation="Check the branch name or your permissions.") - - return MCPResponse( - status=MCPToolStatus.SUCCESS, - data={ - "name": branch.name, - "id": branch.id, - }, - ) - - -@mcp.tool(tags=["branches", "retrieve"], annotations=ToolAnnotations(readOnlyHint=True)) -async def get_branches(ctx: Context) -> MCPResponse[dict[str, BranchData]]: - """Retrieve all branches from infrahub.""" - - client: InfrahubClient = ctx.request_context.lifespan_context.client - ctx.info("Fetching all branches from Infrahub...") - - branches: dict[str, BranchData] = await client.branch.all() - - return MCPResponse(status=MCPToolStatus.SUCCESS, data=branches) diff --git a/src/infrahub_mcp_server/constants.py b/src/infrahub_mcp_server/constants.py deleted file mode 100644 index 01dde86..0000000 --- a/src/infrahub_mcp_server/constants.py +++ /dev/null @@ -1,9 +0,0 @@ -NAMESPACES_INTERNAL = ["Internal", "Profile", "Template"] - -schema_attribute_type_mapping = { - "Text": "String", - "Number": "Integer", - "Boolean": "Boolean", - "DateTime": "DateTime", - "Enum": "String", -} diff --git a/src/infrahub_mcp_server/gql.py b/src/infrahub_mcp_server/gql.py deleted file mode 100644 index 61bb0df..0000000 --- a/src/infrahub_mcp_server/gql.py +++ /dev/null @@ -1,44 +0,0 @@ -from typing import TYPE_CHECKING, Annotated, Any - -from fastmcp import Context, FastMCP -from mcp.types import ToolAnnotations -from pydantic import Field - -from infrahub_mcp_server.utils import MCPResponse, MCPToolStatus - -if TYPE_CHECKING: - from infrahub_sdk import InfrahubClient - -mcp: FastMCP = FastMCP(name="Infrahub GraphQL") - - -@mcp.tool(tags=["schemas", "retrieve"], annotations=ToolAnnotations(readOnlyHint=True)) -async def get_graphql_schema(ctx: Context) -> MCPResponse[str]: - """Retrieve the GraphQL schema from Infrahub - - Parameters: - None - - Returns: - MCPResponse with the GraphQL schema as a string. - """ - client: InfrahubClient = ctx.request_context.lifespan_context.client - resp = await client._get(url=f"{client.address}/schema.graphql") # noqa: SLF001 - return MCPResponse(status=MCPToolStatus.SUCCESS, data=resp.text) - - -@mcp.tool(tags=["schemas", "retrieve"], annotations=ToolAnnotations(readOnlyHint=False)) -async def query_graphql( - ctx: Context, query: Annotated[str, Field(description="GraphQL query to execute.")] -) -> MCPResponse[dict[str, Any]]: - """Execute a GraphQL query against Infrahub. - - Parameters: - query: GraphQL query to execute. - - Returns: - MCPResponse with the result of the query. - - """ - client: InfrahubClient = ctx.request_context.lifespan_context.client - return await client.execute_graphql(query=query) diff --git a/src/infrahub_mcp_server/nodes.py b/src/infrahub_mcp_server/nodes.py deleted file mode 100644 index 4c08fd1..0000000 --- a/src/infrahub_mcp_server/nodes.py +++ /dev/null @@ -1,230 +0,0 @@ -from typing import TYPE_CHECKING, Annotated, Any - -from fastmcp import Context, FastMCP -from infrahub_sdk.exceptions import GraphQLError, SchemaNotFoundError -from infrahub_sdk.types import Order -from mcp.types import ToolAnnotations -from pydantic import Field - -from infrahub_mcp_server.constants import schema_attribute_type_mapping -from infrahub_mcp_server.utils import MCPResponse, MCPToolStatus, _log_and_return_error, convert_node_to_dict - -if TYPE_CHECKING: - from infrahub_sdk.client import InfrahubClient - -mcp: FastMCP = FastMCP(name="Infrahub Nodes") - - -@mcp.tool(tags=["nodes", "retrieve"], annotations=ToolAnnotations(readOnlyHint=True)) -async def get_nodes( - ctx: Context, - kind: Annotated[str, Field(description="Kind of the objects to retrieve.")], - branch: Annotated[ - str | None, - Field(default=None, description="Branch to retrieve the objects from. Defaults to None (uses default branch)."), - ], - filters: Annotated[dict[str, Any] | None, Field(default=None, description="Dictionary of filters to apply.")], - partial_match: Annotated[bool, Field(default=False, description="Whether to use partial matching for filters.")], -) -> MCPResponse[list[str]]: - """Get all objects of a specific kind from Infrahub. - - To retrieve the list of available kinds, use the `get_schema_mapping` tool. - To retrieve the list of available filters for a specific kind, use the `get_node_filters` tool. - - Parameters: - kind: Kind of the objects to retrieve. - branch: Branch to retrieve the objects from. Defaults to None (uses default branch). - filters: Dictionary of filters to apply. - partial_match: Whether to use partial matching for filters. - - Returns: - MCPResponse with success status and objects. - - """ - client: InfrahubClient = ctx.request_context.lifespan_context.client - ctx.info(f"Fetching nodes of kind: {kind} with filters: {filters} from Infrahub...") - - # Verify if the kind exists in the schema and guide Tool if not - try: - schema = await client.schema.get(kind=kind, branch=branch) - except SchemaNotFoundError: - error_msg = f"Schema not found for kind: {kind}." - remediation_msg = "Use the `get_schema_mapping` tool to list available kinds." - return _log_and_return_error(ctx=ctx, error=error_msg, remediation=remediation_msg) - - # TODO: Verify if the filters are valid for the kind and guide Tool if not - - try: - if filters: - nodes = await client.filters( - kind=schema.kind, - branch=branch, - partial_match=partial_match, - parallel=True, - order=Order(disable=True), - populate_store=True, - prefetch_relationships=True, - **filters, - ) - else: - nodes = await client.all( - kind=schema.kind, - branch=branch, - parallel=True, - order=Order(disable=True), - populate_store=True, - prefetch_relationships=True, - ) - except GraphQLError as exc: - return _log_and_return_error(ctx=ctx, error=exc, remediation="Check the provided filters or the kind name.") - - # Format the response with serializable data - # serialized_nodes = [] - # for node in nodes: - # node_data = await convert_node_to_dict(obj=node, branch=branch) - # serialized_nodes.append(node_data) - serialized_nodes = [obj.display_label for obj in nodes] - - # Return the serialized response - ctx.debug(f"Retrieved {len(serialized_nodes)} nodes of kind {kind}") - - return MCPResponse( - status=MCPToolStatus.SUCCESS, - data=serialized_nodes, - ) - - -@mcp.tool(tags=["nodes", "filters", "retrieve"], annotations=ToolAnnotations(readOnlyHint=True)) -async def get_node_filters( - ctx: Context, - kind: Annotated[str, Field(description="Kind of the objects to retrieve.")], - branch: Annotated[ - str | None, - Field(default=None, description="Branch to retrieve the objects from. Defaults to None (uses default branch)."), - ], -) -> MCPResponse[dict[str, str]]: - """Retrieve all the available filters for a specific schema node kind. - - There's multiple types of filters - attribute filters are in the form attribute__value - - relationship filters are in the form relationship__attribute__value - you can find more information on the peer node of the relationship using the `get_schema` tool - - Filters that start with parent refer to a related generic schema node. - You can find the type of that related node by inspected the output of the `get_schema` tool. - - Parameters: - kind: Kind of the objects to retrieve. - branch: Branch to retrieve the objects from. Defaults to None (uses default branch). - - Returns: - MCPResponse with success status and filters. - """ - client: InfrahubClient = ctx.request_context.lifespan_context.client - ctx.info(f"Fetching available filters for kind: {kind} from Infrahub...") - - # Verify if the kind exists in the schema and guide Tool if not - try: - schema = await client.schema.get(kind=kind, branch=branch) - except SchemaNotFoundError: - error_msg = f"Schema not found for kind: {kind}." - remediation_msg = "Use the `get_schema_mapping` tool to list available kinds." - return _log_and_return_error(ctx=ctx, error=error_msg, remediation=remediation_msg) - - filters = { - f"{attribute.name}__value": schema_attribute_type_mapping.get(attribute.kind, "String") - for attribute in schema.attributes - } - - for relationship in schema.relationships: - relationship_schema = await client.schema.get(kind=relationship.peer) - relationship_filters = { - f"{relationship.name}__{attribute.name}__value": schema_attribute_type_mapping.get(attribute.kind, "String") - for attribute in relationship_schema.attributes - } - filters.update(relationship_filters) - - return MCPResponse( - status=MCPToolStatus.SUCCESS, - data=filters, - ) - - -@mcp.tool(tags=["nodes", "retrieve"], annotations=ToolAnnotations(readOnlyHint=True)) -async def get_related_nodes( - ctx: Context, - kind: Annotated[str, Field(description="Kind of the objects to retrieve.")], - relation: Annotated[str, Field(description="Name of the relation to fetch.")], - filters: Annotated[dict[str, Any] | None, Field(default=None, description="Dictionary of filters to apply.")], - branch: Annotated[ - str | None, - Field(default=None, description="Branch to retrieve the objects from. Defaults to None (uses default branch)."), - ], -) -> MCPResponse[list[dict[str, Any]]]: - """Retrieve related nodes by relation name and a kind. - - Args: - kind: Kind of the node to fetch. - filters: Filters to apply on the node to fetch. - relation: Name of the relation to fetch. - branch: Branch to fetch the node from. Defaults to None (uses default branch). - - Returns: - MCPResponse with success status and objects. - - """ - client: InfrahubClient = ctx.request_context.lifespan_context.client - filters = filters or {} - if branch: - ctx.info(f"Fetching nodes related to {kind} with filters {filters} in branch {branch} from Infrahub...") - else: - ctx.info(f"Fetching nodes related to {kind} with filters {filters} from Infrahub...") - - try: - node_id = node_hfid = None - if filters.get("ids"): - node_id = filters["ids"][0] - elif filters.get("hfid"): - node_hfid = filters["hfid"] - if node_id: - node = await client.get( - kind=kind, - id=node_id, - branch=branch, - include=[relation], - prefetch_relationships=True, - populate_store=True, - ) - elif node_hfid: - node = await client.get( - kind=kind, - hfid=node_hfid, - branch=branch, - include=[relation], - prefetch_relationships=True, - populate_store=True, - ) - except Exception as exc: # noqa: BLE001 - return _log_and_return_error(exc) - - rel = getattr(node, relation, None) - if not rel: - _log_and_return_error( - ctx=ctx, - error=f"Relation '{relation}' not found in kind '{kind}'.", - remediation="Check the schema for the kind to confirm if the relation exists.", - ) - peers = [ - await convert_node_to_dict( - branch=branch, - obj=peer.peer, - include_id=True, - ) - for peer in rel.peers - ] - - return MCPResponse( - status=MCPToolStatus.SUCCESS, - data=peers, - ) diff --git a/src/infrahub_mcp_server/prompts/main.md b/src/infrahub_mcp_server/prompts/main.md deleted file mode 100644 index 7218907..0000000 --- a/src/infrahub_mcp_server/prompts/main.md +++ /dev/null @@ -1,8 +0,0 @@ -You are an infrastructure specilist specialized in answering questions about the infrastructure. - -All the information you need are present in Infrahub and you can access it via an MCP server which exposes a number of tools. - -When someone ask about a specific data, you need to: -- Identify what is the associated kind in the schema for this data using the tool `schema_get_mapping` -- Retrieve more information about this specific model, including the option available to filter (tool : `get_node_filters`) -- Use the tool `get_objects` to query one or multiple objects \ No newline at end of file diff --git a/src/infrahub_mcp_server/schema.py b/src/infrahub_mcp_server/schema.py deleted file mode 100644 index 111ebef..0000000 --- a/src/infrahub_mcp_server/schema.py +++ /dev/null @@ -1,140 +0,0 @@ -from typing import TYPE_CHECKING, Annotated, Any - -from fastmcp import Context, FastMCP -from infrahub_sdk.exceptions import BranchNotFoundError, SchemaNotFoundError -from mcp.types import ToolAnnotations -from pydantic import Field - -from infrahub_mcp_server.constants import NAMESPACES_INTERNAL -from infrahub_mcp_server.utils import MCPResponse, MCPToolStatus, _log_and_return_error - -if TYPE_CHECKING: - from infrahub_sdk import InfrahubClient - -mcp: FastMCP = FastMCP(name="Infrahub Schemas") - - -@mcp.tool(tags=["schemas", "retrieve"], annotations=ToolAnnotations(readOnlyHint=True)) -async def get_schema_mapping( - ctx: Context, - branch: Annotated[ - str | None, - Field(default=None, description="Branch to retrieve the mapping from. Defaults to None (uses default branch)."), - ], -) -> MCPResponse[dict[str, str]]: - """List all schema nodes and generics available in Infrahub - - Parameters: - branch: Branch to retrieve the mapping from. Defaults to None (uses default branch). - - Returns: - Dictionary with success status and schema mapping. - """ - client: InfrahubClient = ctx.request_context.lifespan_context.client - if branch: - ctx.info(f"Fetching schema mapping for {branch} from Infrahub...") - else: - ctx.info("Fetching schema mapping from Infrahub...") - - try: - all_schemas = await client.schema.all(branch=branch) - except BranchNotFoundError as exc: - return _log_and_return_error(ctx=ctx, error=exc, remediation="Check the branch name or your permissions.") - - # TODO: Should we add the description ? - schema_mapping = { - kind: node.label or "" for kind, node in all_schemas.items() if node.namespace not in NAMESPACES_INTERNAL - } - - return MCPResponse( - status=MCPToolStatus.SUCCESS, - data=schema_mapping, - ) - - -@mcp.tool(tags=["schemas", "retrieve"], annotations=ToolAnnotations(readOnlyHint=True)) -async def get_schema( - ctx: Context, - kind: Annotated[str, Field(description="Schema Kind to retrieve.")], - branch: Annotated[ - str | None, - Field(default=None, description="Branch to retrieve the schema from. Defaults to None (uses default branch)."), - ], -) -> MCPResponse[dict[str, Any]]: - """Retrieve the full schema for a specific kind. - This includes attributes, relationships, and their types. - - Parameters: - kind: Schema Kind to retrieve. - branch: Branch to retrieve the schema from. Defaults to None (uses default branch). - - Returns: - Dictionary with success status and schema. - """ - client: InfrahubClient = ctx.request_context.lifespan_context.client - ctx.info(f"Fetching schema of {kind} from Infrahub...") - - try: - schema = await client.schema.get(kind=kind, branch=branch) - except SchemaNotFoundError: - error_msg = f"Schema not found for kind: {kind}." - remediation_msg = "Use the `get_schema_mapping` tool to list available kinds." - return _log_and_return_error(ctx=ctx, error=error_msg, remediation=remediation_msg) - except BranchNotFoundError as exc: - return _log_and_return_error(ctx=ctx, error=exc, remediation="Check the branch name or your permissions.") - - schema = await client.schema.get(kind=kind, branch=branch) - - return MCPResponse( - status=MCPToolStatus.SUCCESS, - data=schema.model_dump(), - ) - - -@mcp.tool(tags=["schemas", "retrieve"], annotations=ToolAnnotations(readOnlyHint=True)) -async def get_schemas( - ctx: Context, - branch: Annotated[ - str | None, - Field(default=None, description="Branch to retrieve schemas from. Defaults to None (uses default branch)."), - ], - exclude_profiles: Annotated[ - bool, Field(default=True, description="Whether to exclude Profile schemas. Defaults to True.") - ], - exclude_templates: Annotated[ - bool, Field(default=True, description="Whether to exclude Template schemas. Defaults to True.") - ], -) -> MCPResponse[dict[str, dict[str, Any]]]: - """Retrieve all schemas from Infrahub, optionally excluding Profiles and Templates. - - Parameters: - infrahub_client: Infrahub client to use - branch: Branch to retrieve schemas from - exclude_profiles: Whether to exclude Profile schemas. Defaults to True. - exclude_templates: Whether to exclude Template schemas. Defaults to True. - - Returns: - Dictionary with success status and schemas. - - """ - client: InfrahubClient = ctx.request_context.lifespan_context.client - ctx.info(f"Fetching all schemas in branch {branch or 'main'} from Infrahub...") - - try: - all_schemas = await client.schema.all(branch=branch) - except BranchNotFoundError as exc: - return _log_and_return_error(ctx=ctx, error=exc, remediation="Check the branch name or your permissions.") - - # Filter out Profile and Template if requested - filtered_schemas = {} - for kind, schema in all_schemas.items(): - if (exclude_templates and schema.namespace == "Template") or ( - exclude_profiles and schema.namespace == "Profile" - ): - continue - filtered_schemas[kind] = schema.model_dump() - - return MCPResponse( - status=MCPToolStatus.SUCCESS, - data=filtered_schemas, - ) diff --git a/src/infrahub_mcp_server/server.py b/src/infrahub_mcp_server/server.py deleted file mode 100644 index cf410b9..0000000 --- a/src/infrahub_mcp_server/server.py +++ /dev/null @@ -1,35 +0,0 @@ -from collections.abc import AsyncIterator -from contextlib import asynccontextmanager -from dataclasses import dataclass - -from fastmcp import FastMCP -from infrahub_sdk.client import InfrahubClient - -from infrahub_mcp_server.branch import mcp as branch_mcp -from infrahub_mcp_server.gql import mcp as graphql_mcp -from infrahub_mcp_server.nodes import mcp as nodes_mcp -from infrahub_mcp_server.schema import mcp as schema_mcp - - -@dataclass -class AppContext: - client: InfrahubClient - - -@asynccontextmanager -async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]: # noqa: ARG001, RUF029 - """Manages application lifecycle with type-safe context for the FastMCP server.""" - client = InfrahubClient() - try: - yield AppContext(client=client) - finally: - pass - - -mcp: FastMCP = FastMCP(name="Infrahub MCP Server", version="0.1.0", lifespan=app_lifespan) - -# Mount the various MCPs to the main server -mcp.mount(branch_mcp) -mcp.mount(graphql_mcp) -mcp.mount(nodes_mcp) -mcp.mount(schema_mcp) diff --git a/src/infrahub_mcp_server/utils.py b/src/infrahub_mcp_server/utils.py deleted file mode 100644 index 35b27e6..0000000 --- a/src/infrahub_mcp_server/utils.py +++ /dev/null @@ -1,94 +0,0 @@ -from enum import Enum -from pathlib import Path -from typing import Any, TypeVar - -from fastmcp import Context -from infrahub_sdk.node import Attribute, InfrahubNode, RelatedNode, RelationshipManager -from pydantic import BaseModel - -CURRENT_DIRECTORY = Path(__file__).parent.resolve() -PROMPTS_DIRECTORY = CURRENT_DIRECTORY / "prompts" - - -T = TypeVar("T") - - -class MCPToolStatus(Enum): - SUCCESS = "success" - ERROR = "error" - - -class MCPResponse[T](BaseModel): - status: MCPToolStatus - data: T | None = None - error: str | None = None - remediation: str | None = None - - -def get_prompt(name: str) -> str: - prompt_file = PROMPTS_DIRECTORY / f"{name}.md" - if not prompt_file.exists(): - raise FileNotFoundError(f"Prompt file '{prompt_file}' does not exist.") - return (PROMPTS_DIRECTORY / f"{name}.md").read_text() - - -def _log_and_return_error(ctx: Context, error: str | Exception, remediation: str | None = None) -> MCPResponse: - """Log an error and return a standardized error response.""" - if isinstance(error, Exception): - error = str(error) - ctx.error(message=error) - return MCPResponse( - status=MCPToolStatus.ERROR, - error=error, - remediation=remediation, - ) - - -async def convert_node_to_dict(*, obj: InfrahubNode, branch: str | None, include_id: bool = True) -> dict[str, Any]: # noqa: C901 - data = {} - - if include_id: - data["index"] = obj.id or None - - for attr_name in obj._schema.attribute_names: # noqa: SLF001 - attr: Attribute = getattr(obj, attr_name) - data[attr_name] = str(attr.value) - - for rel_name in obj._schema.relationship_names: # noqa: SLF001 - rel = getattr(obj, rel_name) - if rel and isinstance(rel, RelatedNode): - if not rel.initialized: - await rel.fetch() - related_node = obj._client.store.get( # noqa: SLF001 - branch=branch, - key=rel.peer.id, - raise_when_missing=False, - ) - if related_node: - data[rel_name] = ( - related_node.get_human_friendly_id_as_string(include_kind=True) - if related_node.hfid - else related_node.id - ) - elif rel and isinstance(rel, RelationshipManager): - peers: list[dict[str, Any]] = [] - if not rel.initialized: - await rel.fetch() - for peer in rel.peers: - # FIXME: We are using the store to avoid doing to many queries to Infrahub - # but we could end up doing store+infrahub if the store is not populated - related_node = obj._client.store.get( # noqa: SLF001 - key=peer.id, - raise_when_missing=False, - branch=branch, - ) - if not related_node: - await peer.fetch() - related_node = peer.peer - peers.append( - related_node.get_human_friendly_id_as_string(include_kind=True) - if related_node.hfid - else related_node.id, - ) - data[rel_name] = peers - return data