Open Policy Agent (OPA) REST API MCP Server

# generated by fastapi-codegen: # filename: openapi.yaml # timestamp: 2025-06-29T07:58:39+00:00 import argparse import json import os from typing import * from typing import Optional, Union from autogen.mcp.mcp_proxy import MCPProxy from autogen.mcp.mcp_proxy.security import BaseSecurity from fastapi import Header from models import ( DataSchema, Field200Result, Field200SingleResult, Field400, Field404, Input, PartialQuerySchema, PatchesSchema, QueryInputSchema, V0DataPathPostResponse, V1CompilePostResponse, V1DataPathGetResponse, V1DataPathPostResponse, V1PoliciesIdDeleteResponse, V1QueryGetResponse, V1QueryPostResponse, ) app = MCPProxy( contact={'name': 'The OPA team', 'url': 'https://github.com/open-policy-agent/opa'}, description='OPA provides policy-based control for cloud native environments. The following *endpoints* (such as `PUT /v1/policies`) provide reference documentation for the OPA REST API.\n\n### API specification viewing options\n\n- **[View the specification in *Redoc* (default)](index.html)**\n- **[View the specification in *Swagger UI*](swagger-ui.html)**', license={ 'name': 'Apache 2.0', 'url': 'https://www.apache.org/licenses/LICENSE-2.0', }, title='Open Policy Agent (OPA) REST API', version='0.28.0', servers=[{'url': 'http://openpolicy.local'}], ) @app.post( '/#-datamodel-code-generator-#-root-#-special-#', description=""" This API queries the document at */data/system/main* by default (however, you can [configure OPA](https://www.openpolicyagent.org/docs/latest/configuration/) to use a different path to serve these queries). That document defines the response. For example, use the following in `PUT /v1/policies/{path}`) to define a rule that will produce a value for the */data/system/main* document: ```yaml package system main = msg { msg := sprintf("hello, %v", input.user) } ``` The server will return a *not found* (404) response if */data/system/main* is undefined. """, tags=['query_execution'], ) def post_simple_query(pretty: Optional[bool] = None, body: QueryInputSchema = ...): """ Execute a simple query """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/health', description=""" This API endpoint verifies that the server is operational. The response from the server is either 200 or 500: - **200** - OPA service is healthy. If `bundles` is true, then all configured bundles have been activated. If `plugins` is true, then all plugins are in an 'OK' state. - **500** - OPA service is *not* healthy. If `bundles` is true, at least one of configured bundles has not yet been activated. If `plugins` is true, at least one plugins is in a 'not OK' state. --- **Note** This check is only for initial bundle activation. Subsequent downloads will not affect the health check. Use the **status** endpoint (in the (management API)[management.html]) for more fine-grained bundle status monitoring. --- """, tags=['system_health_check'], ) def get_health(bundles: Optional[bool] = None, plugins: Optional[bool] = None): """ Health """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/v0/data/{path}', description=""" The example given here assumes you have created a policy (with `PUT /v1/policies/{path}`), such as: ```yaml package opa.examples import input.example.flag allow_request { flag == true } ``` The server will return a *not found* (404) response if the requested document is missing or undefined. """, tags=['document_management', 'query_execution'], ) def get_document_with_web_hook(pretty: Optional[bool] = None, path: str = ...): """ Get a document (with webhook) """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/v1/compile', description=""" This API endpoint allows you to partially evaluate Rego queries and obtain a simplified version of the policy. The example below assumes that OPA has been given the following policy (use `PUT /v1/policies/{path}`): ```yaml package example allow { input.subject.clearance_level >= data.reports[_].clearance_level } ``` Compile API **request body** so that it contain the following fields: | Field | Type | Required | Description | | --- | --- | --- | --- | | `query` | `string` | Yes | The query to partially evaluate and compile. | | `input` | `any` | No | The input document to use during partial evaluation (default: undefined). | | `unknowns` | `array[string]` | No | The terms to treat as unknown during partial evaluation (default: `["input"]`]). | For example: ```json { "query": "data.example.allow == true", "input": { "subject": { "clearance_level": 4 } }, "unknowns": [ "data.reports" ] } ``` ### Partial evaluation In some cases, the result of partial valuation is a conclusive, unconditional answer. See [the guidance](https://www.openpolicyagent.org/docs/latest/rest-api/#unconditional-results-from-partial-evaluation) for details. """, tags=['query_execution'], ) def post_compile( pretty: Optional[bool] = None, explain: Optional[str] = None, metrics: Optional[bool] = None, instrument: Optional[bool] = None, body: PartialQuerySchema = None, ): """ Compile """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/v1/config', description=""" This API endpoint responds with active configuration (result response) --- **Note** The `credentials` field in the `services` configuration and The `private_key` and `key` fields in the `keys` configuration will be omitted from the API response --- """, tags=['system_configuration_retrieval'], ) def get_config(pretty: Optional[bool] = None): """ Get configurations """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/v1/data/{path}', description=""" This API endpoint deletes an existing document from the server """, tags=['document_management'], ) def delete_document(path: str): """ Delete a document """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/v1/data/{path}', description=""" This API endpoint returns the document specified by `path`. The server will return a *bad request* (400) response if either: - The query requires an input document and you do not provide it - You provide the input document but the query has already defined it. """, tags=['document_management', 'query_execution'], ) def get_document( input: Optional[Input] = None, pretty: Optional[bool] = None, provenance: Optional[bool] = None, explain: Optional[str] = None, metrics: Optional[bool] = None, instrument: Optional[bool] = None, path: str = ..., ): """ Get a document """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.patch( '/v1/data/{path}', description=""" This API endpoint updates an existing document on the server by describing the changes required (using [JSON patch operations](http://jsonpatch.com/)) """, tags=['document_management'], ) def patch_document(path: str, body: PatchesSchema = ...): """ Update a document """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/v1/data/{path}', description=""" The server will return a *bad request* (400) response if either: - The query requires an input document and you do not provide it - You provided an input document but the query has already defined it. If `path` indexes into an array, the server will attempt to convert the array index to an integer. If the path element cannot be converted to an integer, a *not found* response (404) will be returned. """, tags=['document_management', 'query_execution'], ) def get_document_with_path( pretty: Optional[bool] = None, provenance: Optional[bool] = None, explain: Optional[str] = None, metrics: Optional[bool] = None, instrument: Optional[bool] = None, path: str = ..., ): """ Get a document (with input) """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.put( '/v1/data/{path}', description=""" If the path does not refer to an existing document (for example *us-west/servers*), the server will attempt to create all the necessary containing documents. This behavior is similar to the Unix command [mkdir -p](https://en.wikipedia.org/wiki/Mkdir#Options). """, tags=['document_management'], ) def put_document( if__none__match: Optional[str] = Header(None, alias='If-None-Match'), path: str = ..., body: DataSchema = ..., ): """ Create or overwrite a document """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/v1/policies', description=""" This API endpoint responds with a list of all policy modules on the server (result response) """, tags=['policy_module_management'], ) def get_policies(pretty: Optional[bool] = None): """ List policies """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/v1/policies/{id}', description=""" This API endpoint removes an existing policy module from the server """, tags=['policy_module_management', 'document_management'], ) def delete_policy_module(pretty: Optional[bool] = None, id: str = ...): """ Delete a policy module """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/v1/policies/{id}', description=""" This API endpoint returns the details of the specified policy module (`{id}`) """, tags=['policy_module_management'], ) def get_policy_module(pretty: Optional[bool] = None, id: str = ...): """ Get a policy module """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.put( '/v1/policies/{id}', description=""" - If the policy module does not exist, it is created. - If the policy module already exists, it is replaced. If the policy module isn't correctly defined, a *bad request* (400) response is returned. ### Example policy module ```yaml package opa.examples import data.servers import data.networks import data.ports public_servers[server] { some k, m server := servers[_] server.ports[_] == ports[k].id ports[k].networks[_] == networks[m].id networks[m].public == true } ``` """, tags=['policy_module_management'], ) def put_policy_module( pretty: Optional[bool] = None, metrics: Optional[bool] = None, id: str = ... ): """ Create or update a policy module """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/v1/query', description=""" This API endpoint returns bindings for the variables in the query. For more complex JSON queries, use `POST /v1/query` instead. """, tags=['query_execution'], ) def get_query( q: str, pretty: Optional[bool] = None, explain: Optional[str] = None, metrics: Optional[bool] = None, ): """ Execute an ad-hoc query (simple) """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/v1/query', description=""" This API endpoint returns bindings for the variables in the query. For simpler JSON queries, you may use `GET /v1/query` instead. """, tags=['query_execution'], ) def post_query( pretty: Optional[bool] = None, explain: Optional[str] = None, metrics: Optional[bool] = None, ): """ Execute an ad-hoc query (complex) """ raise RuntimeError("Should be patched by MCPProxy and never executed") if __name__ == "__main__": parser = argparse.ArgumentParser(description="MCP Server") parser.add_argument( "transport", choices=["stdio", "sse", "streamable-http"], help="Transport mode (stdio, sse or streamable-http)", ) args = parser.parse_args() if "CONFIG_PATH" in os.environ: config_path = os.environ["CONFIG_PATH"] app.load_configuration(config_path) if "CONFIG" in os.environ: config = os.environ["CONFIG"] app.load_configuration_from_string(config) if "SECURITY" in os.environ: security_params = BaseSecurity.parse_security_parameters_from_env( os.environ, ) app.set_security_params(security_params) mcp_settings = json.loads(os.environ.get("MCP_SETTINGS", "{}")) app.get_mcp(**mcp_settings).run(transport=args.transport)