InfluxDB OSS API MCP Server

# generated by fastapi-codegen: # filename: openapi.yaml # timestamp: 2025-06-29T03:45:53+00:00 import argparse import json import os from datetime import datetime from typing import * from typing import Optional, Union from autogen.mcp.mcp_proxy import MCPProxy from autogen.mcp.mcp_proxy.security import BaseSecurity, HTTPBasic from fastapi import Header, Path, Query from pydantic import conint, constr from models import ( DBRP, Accept, Accept1, AcceptEncoding, AddResourceMemberRequestBody, AnalyzeQueryResponse, ASTResponse, Authorization, AuthorizationPostRequest, Authorizations, AuthorizationUpdateRequest, Bucket, Buckets, Cell, Cells, CellUpdate, Check, CheckPatch, Checks, ContentEncoding, ContentType, ContentType1, ContentType3, CreateCell, CreateDashboardRequest, Dashboard, Dashboards, DashboardsDashboardIDGetResponse, DashboardsDashboardIDPatchRequest, DashboardsPostResponse, DBRPs, DBRPUpdate, DeletePredicateRequest, Document, DocumentCreate, Documents, DocumentUpdate, Error1, Flags, FluxResponse, FluxSuggestion, FluxSuggestions, HealthCheck, Id, Include, IsOnboarding, LabelCreateRequest, LabelMapping, LabelResponse, LabelsResponse, LabelUpdate, LanguageRequest, LineProtocolError, LineProtocolLengthError, Logs, NotificationEndpoint, NotificationEndpoints, NotificationEndpointUpdate, NotificationRule, NotificationRules, NotificationRuleUpdate, OnboardingRequest, OnboardingResponse, Organization, Organizations, PasswordResetBody, PatchBucketRequest, PatchOrganizationRequest, PostBucketRequest, PostCheck, PostNotificationEndpoint, PostNotificationRule, PostOrganizationRequest, Query, QueryPostRequest, Ready, ResourceMember, ResourceMembers, ResourceOwner, ResourceOwners, Routes, Run, RunManually, Runs, ScraperTargetRequest, ScraperTargetResponse, ScraperTargetResponses, SecretKeys, SecretKeysResponse, Secrets, SortBy, Source, Sources, Stack, StacksGetResponse, StacksPostRequest, StacksStackIdPatchRequest, Status9, Task, TaskCreateRequest, Tasks, TaskUpdateRequest, Telegraf, TelegrafPlugins, TelegrafRequest, Telegrafs, Template, TemplateApply, TemplatesExportPostRequest, TemplateSummary, User, UserResponse, Users, Variable2, Variables, View, WritePrecision, ) app = MCPProxy( description='# Authentication\n\n<!-- ReDoc-Inject: <security-definitions> -->', title='Influx OSS API Service', version='2.0.0', servers=[{'url': '/api/v2'}], ) @app.get('/#-datamodel-code-generator-#-root-#-special-#', tags=['route_management']) def get_routes(zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span')): """ List all top level routes """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/authorizations', tags=['authorization_handling']) def get_authorizations( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), user_i_d: Optional[str] = Query(None, alias='userID'), user: Optional[str] = None, org_i_d: Optional[str] = Query(None, alias='orgID'), org: Optional[str] = None, ): """ List all authorizations """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/authorizations', tags=['authorization_handling']) def post_authorizations( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), body: AuthorizationPostRequest = ..., ): """ Create an authorization """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete('/authorizations/{authID}', tags=['authorization_handling']) def delete_authorizations_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), auth_i_d: str = Path(..., alias='authID'), ): """ Delete an authorization """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/authorizations/{authID}', tags=['authorization_handling']) def get_authorizations_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), auth_i_d: str = Path(..., alias='authID'), ): """ Retrieve an authorization """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.patch('/authorizations/{authID}', tags=['authorization_handling']) def patch_authorizations_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), auth_i_d: str = Path(..., alias='authID'), body: AuthorizationUpdateRequest = ..., ): """ Update an authorization to be active or inactive """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/buckets', tags=['bucket_operations']) def get_buckets( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), offset: Optional[conint(ge=0)] = None, limit: Optional[conint(ge=1, le=100)] = 20, after: Optional[str] = None, org: Optional[str] = None, org_i_d: Optional[str] = Query(None, alias='orgID'), name: Optional[str] = None, id: Optional[str] = None, ): """ List all buckets """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/buckets', tags=['bucket_operations']) def post_buckets( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), body: PostBucketRequest = ..., ): """ Create a bucket """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete('/buckets/{bucketID}', tags=['bucket_operations']) def delete_buckets_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), bucket_i_d: str = Path(..., alias='bucketID'), ): """ Delete a bucket """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/buckets/{bucketID}', tags=['bucket_operations']) def get_buckets_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), bucket_i_d: str = Path(..., alias='bucketID'), ): """ Retrieve a bucket """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.patch('/buckets/{bucketID}', tags=['bucket_operations']) def patch_buckets_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), bucket_i_d: str = Path(..., alias='bucketID'), body: PatchBucketRequest = ..., ): """ Update a bucket """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/buckets/{bucketID}/labels', tags=['bucket_operations', 'label_operations']) def get_buckets_i_d_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), bucket_i_d: str = Path(..., alias='bucketID'), ): """ List all labels for a bucket """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/buckets/{bucketID}/labels', tags=['bucket_operations', 'label_operations']) def post_buckets_i_d_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), bucket_i_d: str = Path(..., alias='bucketID'), body: LabelMapping = ..., ): """ Add a label to a bucket """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/buckets/{bucketID}/labels/{labelID}', tags=['bucket_operations', 'label_operations'], ) def delete_buckets_i_d_labels_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), bucket_i_d: str = Path(..., alias='bucketID'), label_i_d: str = Path(..., alias='labelID'), ): """ Delete a label from a bucket """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/buckets/{bucketID}/members', tags=['bucket_operations', 'user_account_management'] ) def get_buckets_i_d_members( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), bucket_i_d: str = Path(..., alias='bucketID'), ): """ List all users with member privileges for a bucket """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/buckets/{bucketID}/members', tags=['bucket_operations']) def post_buckets_i_d_members( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), bucket_i_d: str = Path(..., alias='bucketID'), body: AddResourceMemberRequestBody = ..., ): """ Add a member to a bucket """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/buckets/{bucketID}/members/{userID}', tags=['bucket_operations', 'organization_operations', 'scraper_member_handling'], ) def delete_buckets_i_d_members_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), user_i_d: str = Path(..., alias='userID'), bucket_i_d: str = Path(..., alias='bucketID'), ): """ Remove a member from a bucket """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/buckets/{bucketID}/owners', tags=['bucket_operations']) def get_buckets_i_d_owners( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), bucket_i_d: str = Path(..., alias='bucketID'), ): """ List all owners of a bucket """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/buckets/{bucketID}/owners', tags=['bucket_operations', 'authorization_handling'] ) def post_buckets_i_d_owners( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), bucket_i_d: str = Path(..., alias='bucketID'), body: AddResourceMemberRequestBody = ..., ): """ Add an owner to a bucket """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/buckets/{bucketID}/owners/{userID}', tags=['bucket_operations', 'user_account_management'], ) def delete_buckets_i_d_owners_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), user_i_d: str = Path(..., alias='userID'), bucket_i_d: str = Path(..., alias='bucketID'), ): """ Remove an owner from a bucket """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/checks', tags=['check_operations']) def get_checks( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), offset: Optional[conint(ge=0)] = None, limit: Optional[conint(ge=1, le=100)] = 20, org_i_d: str = Query(..., alias='orgID'), ): """ List all checks """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/checks', tags=['check_operations']) def create_check(body: PostCheck): """ Add new check """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete('/checks/{checkID}', tags=['check_operations']) def delete_checks_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), check_i_d: str = Path(..., alias='checkID'), ): """ Delete a check """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/checks/{checkID}', tags=['check_operations']) def get_checks_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), check_i_d: str = Path(..., alias='checkID'), ): """ Retrieve a check """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.patch('/checks/{checkID}', tags=['check_operations']) def patch_checks_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), check_i_d: str = Path(..., alias='checkID'), body: CheckPatch = ..., ): """ Update a check """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.put('/checks/{checkID}', tags=['check_operations']) def put_checks_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), check_i_d: str = Path(..., alias='checkID'), body: Check = ..., ): """ Update a check """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/checks/{checkID}/labels', tags=['check_operations', 'label_operations']) def get_checks_i_d_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), check_i_d: str = Path(..., alias='checkID'), ): """ List all labels for a check """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/checks/{checkID}/labels', tags=['check_operations', 'label_operations']) def post_checks_i_d_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), check_i_d: str = Path(..., alias='checkID'), body: LabelMapping = ..., ): """ Add a label to a check """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/checks/{checkID}/labels/{labelID}', tags=['check_operations', 'label_operations'] ) def delete_checks_i_d_labels_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), check_i_d: str = Path(..., alias='checkID'), label_i_d: str = Path(..., alias='labelID'), ): """ Delete label from a check """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/checks/{checkID}/query', tags=['check_operations']) def get_checks_i_d_query( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), check_i_d: str = Path(..., alias='checkID'), ): """ Retrieve a check query """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/dashboards', tags=['dashboard_management']) def get_dashboards( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), offset: Optional[conint(ge=0)] = None, limit: Optional[conint(ge=1, le=100)] = 20, descending: Optional[bool] = False, owner: Optional[str] = None, sort_by: Optional[SortBy] = Query(None, alias='sortBy'), id: Optional[Id] = None, org_i_d: Optional[str] = Query(None, alias='orgID'), org: Optional[str] = None, ): """ List all dashboards """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/dashboards', tags=['dashboard_management']) def post_dashboards( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), body: CreateDashboardRequest = ..., ): """ Create a dashboard """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete('/dashboards/{dashboardID}', tags=['dashboard_management']) def delete_dashboards_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), dashboard_i_d: str = Path(..., alias='dashboardID'), ): """ Delete a dashboard """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/dashboards/{dashboardID}', tags=['dashboard_management']) def get_dashboards_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), dashboard_i_d: str = Path(..., alias='dashboardID'), include: Optional[Include] = None, ): """ Retrieve a Dashboard """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.patch('/dashboards/{dashboardID}', tags=['dashboard_management']) def patch_dashboards_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), dashboard_i_d: str = Path(..., alias='dashboardID'), body: DashboardsDashboardIDPatchRequest = ..., ): """ Update a dashboard """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/dashboards/{dashboardID}/cells', tags=['dashboard_management']) def post_dashboards_i_d_cells( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), dashboard_i_d: str = Path(..., alias='dashboardID'), body: CreateCell = ..., ): """ Create a dashboard cell """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.put( '/dashboards/{dashboardID}/cells', description=""" Replaces all cells in a dashboard. This is used primarily to update the positional information of all cells. """, tags=['dashboard_management'], ) def put_dashboards_i_d_cells( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), dashboard_i_d: str = Path(..., alias='dashboardID'), body: Cells = ..., ): """ Replace cells in a dashboard """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete('/dashboards/{dashboardID}/cells/{cellID}', tags=['dashboard_management']) def delete_dashboards_i_d_cells_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), dashboard_i_d: str = Path(..., alias='dashboardID'), cell_i_d: str = Path(..., alias='cellID'), ): """ Delete a dashboard cell """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.patch( '/dashboards/{dashboardID}/cells/{cellID}', description=""" Updates the non positional information related to a cell. Updates to a single cell's positional data could cause grid conflicts. """, tags=['dashboard_management'], ) def patch_dashboards_i_d_cells_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), dashboard_i_d: str = Path(..., alias='dashboardID'), cell_i_d: str = Path(..., alias='cellID'), body: CellUpdate = ..., ): """ Update the non-positional information related to a cell """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/dashboards/{dashboardID}/cells/{cellID}/view', tags=['dashboard_management']) def get_dashboards_i_d_cells_i_d_view( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), dashboard_i_d: str = Path(..., alias='dashboardID'), cell_i_d: str = Path(..., alias='cellID'), ): """ Retrieve the view for a cell """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.patch( '/dashboards/{dashboardID}/cells/{cellID}/view', tags=['dashboard_management'] ) def patch_dashboards_i_d_cells_i_d_view( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), dashboard_i_d: str = Path(..., alias='dashboardID'), cell_i_d: str = Path(..., alias='cellID'), body: View = ..., ): """ Update the view for a cell """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/dashboards/{dashboardID}/labels', tags=['dashboard_management', 'label_operations'], ) def get_dashboards_i_d_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), dashboard_i_d: str = Path(..., alias='dashboardID'), ): """ List all labels for a dashboard """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/dashboards/{dashboardID}/labels', tags=['dashboard_management', 'label_operations'], ) def post_dashboards_i_d_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), dashboard_i_d: str = Path(..., alias='dashboardID'), body: LabelMapping = ..., ): """ Add a label to a dashboard """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/dashboards/{dashboardID}/labels/{labelID}', tags=['dashboard_management', 'label_operations'], ) def delete_dashboards_i_d_labels_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), dashboard_i_d: str = Path(..., alias='dashboardID'), label_i_d: str = Path(..., alias='labelID'), ): """ Delete a label from a dashboard """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/dashboards/{dashboardID}/members', tags=['dashboard_management', 'organization_operations', 'scraper_member_handling'], ) def get_dashboards_i_d_members( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), dashboard_i_d: str = Path(..., alias='dashboardID'), ): """ List all dashboard members """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/dashboards/{dashboardID}/members', tags=['dashboard_management', 'user_account_management'], ) def post_dashboards_i_d_members( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), dashboard_i_d: str = Path(..., alias='dashboardID'), body: AddResourceMemberRequestBody = ..., ): """ Add a member to a dashboard """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/dashboards/{dashboardID}/members/{userID}', tags=['dashboard_management', 'user_account_management', 'organization_operations'], ) def delete_dashboards_i_d_members_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), user_i_d: str = Path(..., alias='userID'), dashboard_i_d: str = Path(..., alias='dashboardID'), ): """ Remove a member from a dashboard """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/dashboards/{dashboardID}/owners', tags=['dashboard_management']) def get_dashboards_i_d_owners( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), dashboard_i_d: str = Path(..., alias='dashboardID'), ): """ List all dashboard owners """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/dashboards/{dashboardID}/owners', tags=['dashboard_management', 'user_account_management'], ) def post_dashboards_i_d_owners( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), dashboard_i_d: str = Path(..., alias='dashboardID'), body: AddResourceMemberRequestBody = ..., ): """ Add an owner to a dashboard """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/dashboards/{dashboardID}/owners/{userID}', tags=['dashboard_management', 'user_account_management'], ) def delete_dashboards_i_d_owners_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), user_i_d: str = Path(..., alias='userID'), dashboard_i_d: str = Path(..., alias='dashboardID'), ): """ Remove an owner from a dashboard """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/dbrps', tags=['retention_policy_management']) def get_d_b_r_ps( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), org_i_d: str = Query(..., alias='orgID'), id: Optional[str] = None, bucket_i_d: Optional[str] = Query(None, alias='bucketID'), default: Optional[bool] = None, db: Optional[str] = None, rp: Optional[str] = None, ): """ List all database retention policy mappings """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/dbrps', tags=['retention_policy_management']) def post_d_b_r_p( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), body: DBRP = ..., ): """ Add a database retention policy mapping """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete('/dbrps/{dbrpID}', tags=['retention_policy_management']) def delete_d_b_r_p_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), org_i_d: str = Query(..., alias='orgID'), dbrp_i_d: str = Path(..., alias='dbrpID'), ): """ Delete a database retention policy """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/dbrps/{dbrpID}', tags=['retention_policy_management']) def get_d_b_r_ps_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), org_i_d: str = Query(..., alias='orgID'), dbrp_i_d: str = Path(..., alias='dbrpID'), ): """ Retrieve a database retention policy mapping """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.patch('/dbrps/{dbrpID}', tags=['retention_policy_management']) def patch_d_b_r_p_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), org_i_d: str = Query(..., alias='orgID'), dbrp_i_d: str = Path(..., alias='dbrpID'), body: DBRPUpdate = ..., ): """ Update a database retention policy mapping """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/delete', tags=['bucket_operations', 'influxdb_query_handling']) def post_delete( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), org: Optional[str] = None, bucket: Optional[str] = None, org_i_d: Optional[str] = Query(None, alias='orgID'), bucket_i_d: Optional[str] = Query(None, alias='bucketID'), body: DeletePredicateRequest = ..., ): """ Delete time series data from InfluxDB """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/documents/templates', tags=['document_template_operations', 'template_operations'] ) def get_documents_templates( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), org: Optional[str] = None, org_i_d: Optional[str] = Query(None, alias='orgID'), ): """ List all templates """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/documents/templates', tags=['document_template_operations', 'template_operations'] ) def post_documents_templates( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), body: DocumentCreate = ..., ): """ Create a template """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/documents/templates/{templateID}', tags=['document_template_operations', 'template_operations'], ) def delete_documents_templates_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), template_i_d: str = Path(..., alias='templateID'), ): """ Delete a template """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/documents/templates/{templateID}', tags=['document_template_operations', 'template_operations'], ) def get_documents_templates_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), template_i_d: str = Path(..., alias='templateID'), ): """ Retrieve a template """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.put( '/documents/templates/{templateID}', tags=['template_operations', 'document_template_operations'], ) def put_documents_templates_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), template_i_d: str = Path(..., alias='templateID'), body: DocumentUpdate = ..., ): """ Update a template """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/documents/templates/{templateID}/labels', tags=['template_operations', 'label_operations'], ) def get_documents_templates_i_d_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), template_i_d: str = Path(..., alias='templateID'), ): """ List all labels for a template """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/documents/templates/{templateID}/labels', tags=['template_operations', 'label_operations'], ) def post_documents_templates_i_d_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), template_i_d: str = Path(..., alias='templateID'), body: LabelMapping = ..., ): """ Add a label to a template """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/documents/templates/{templateID}/labels/{labelID}', tags=['template_operations', 'label_operations', 'document_template_operations'], ) def delete_documents_templates_i_d_labels_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), template_i_d: str = Path(..., alias='templateID'), label_i_d: str = Path(..., alias='labelID'), ): """ Delete a label from a template """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/flags', tags=['system_health_check', 'user_account_management']) def get_flags(zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span')): """ Return the feature flags for the currently authenticated user """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/health', tags=['system_health_check']) def get_health(zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span')): """ Get the health of an instance """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/labels', tags=['label_operations']) def get_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), org_i_d: Optional[str] = Query(None, alias='orgID'), ): """ List all labels """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/labels', tags=['label_operations']) def post_labels(body: LabelCreateRequest): """ Create a label """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete('/labels/{labelID}', tags=['label_operations']) def delete_labels_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), label_i_d: str = Path(..., alias='labelID'), ): """ Delete a label """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/labels/{labelID}', tags=['label_operations']) def get_labels_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), label_i_d: str = Path(..., alias='labelID'), ): """ Retrieve a label """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.patch('/labels/{labelID}', tags=['label_operations', 'bucket_operations']) def patch_labels_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), label_i_d: str = Path(..., alias='labelID'), body: LabelUpdate = ..., ): """ Update a label """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/me', tags=[ 'user_account_management', 'user_session_operations', 'authorization_handling', 'system_health_check', ], ) def get_me(zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span')): """ Retrieve the currently authenticated user """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.put( '/me/password', tags=['user_account_management', 'authorization_handling'], security=[ HTTPBasic(name="None"), ], ) def put_me_password( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), body: PasswordResetBody = ..., ): """ Update a password """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/notificationEndpoints', tags=['notification_endpoint_handling']) def get_notification_endpoints( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), offset: Optional[conint(ge=0)] = None, limit: Optional[conint(ge=1, le=100)] = 20, org_i_d: str = Query(..., alias='orgID'), ): """ List all notification endpoints """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/notificationEndpoints', tags=['notification_endpoint_handling', 'label_operations'], ) def create_notification_endpoint(body: PostNotificationEndpoint): """ Add a notification endpoint """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/notificationEndpoints/{endpointID}', tags=['notification_endpoint_handling', 'label_operations'], ) def delete_notification_endpoints_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), endpoint_i_d: str = Path(..., alias='endpointID'), ): """ Delete a notification endpoint """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/notificationEndpoints/{endpointID}', tags=['notification_endpoint_handling', 'label_operations'], ) def get_notification_endpoints_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), endpoint_i_d: str = Path(..., alias='endpointID'), ): """ Retrieve a notification endpoint """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.patch( '/notificationEndpoints/{endpointID}', tags=['notification_endpoint_handling', 'label_operations'], ) def patch_notification_endpoints_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), endpoint_i_d: str = Path(..., alias='endpointID'), body: NotificationEndpointUpdate = ..., ): """ Update a notification endpoint """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.put( '/notificationEndpoints/{endpointID}', tags=['notification_endpoint_handling', 'label_operations'], ) def put_notification_endpoints_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), endpoint_i_d: str = Path(..., alias='endpointID'), body: NotificationEndpoint = ..., ): """ Update a notification endpoint """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/notificationEndpoints/{endpointID}/labels', tags=['label_operations', 'notification_endpoint_handling'], ) def get_notification_endpoints_i_d_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), endpoint_i_d: str = Path(..., alias='endpointID'), ): """ List all labels for a notification endpoint """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/notificationEndpoints/{endpointID}/labels', tags=['label_operations', 'notification_endpoint_handling'], ) def post_notification_endpoint_i_d_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), endpoint_i_d: str = Path(..., alias='endpointID'), body: LabelMapping = ..., ): """ Add a label to a notification endpoint """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/notificationEndpoints/{endpointID}/labels/{labelID}', tags=['label_operations', 'notification_endpoint_handling'], ) def delete_notification_endpoints_i_d_labels_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), endpoint_i_d: str = Path(..., alias='endpointID'), label_i_d: str = Path(..., alias='labelID'), ): """ Delete a label from a notification endpoint """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/notificationRules', tags=['notification_rule_handling']) def get_notification_rules( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), offset: Optional[conint(ge=0)] = None, limit: Optional[conint(ge=1, le=100)] = 20, org_i_d: str = Query(..., alias='orgID'), check_i_d: Optional[str] = Query(None, alias='checkID'), tag: Optional[constr(pattern=r'^[a-zA-Z0-9_]+:[a-zA-Z0-9_]+$')] = None, ): """ List all notification rules """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/notificationRules', tags=['notification_rule_handling', 'notification_endpoint_handling'], ) def create_notification_rule(body: PostNotificationRule): """ Add a notification rule """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete('/notificationRules/{ruleID}', tags=['notification_rule_handling']) def delete_notification_rules_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), rule_i_d: str = Path(..., alias='ruleID'), ): """ Delete a notification rule """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/notificationRules/{ruleID}', tags=['notification_rule_handling']) def get_notification_rules_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), rule_i_d: str = Path(..., alias='ruleID'), ): """ Retrieve a notification rule """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.patch( '/notificationRules/{ruleID}', tags=['notification_rule_handling', 'notification_endpoint_handling'], ) def patch_notification_rules_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), rule_i_d: str = Path(..., alias='ruleID'), body: NotificationRuleUpdate = ..., ): """ Update a notification rule """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.put( '/notificationRules/{ruleID}', tags=['notification_rule_handling', 'notification_endpoint_handling'], ) def put_notification_rules_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), rule_i_d: str = Path(..., alias='ruleID'), body: NotificationRule = ..., ): """ Update a notification rule """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/notificationRules/{ruleID}/labels', tags=['label_operations', 'notification_rule_handling'], ) def get_notification_rules_i_d_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), rule_i_d: str = Path(..., alias='ruleID'), ): """ List all labels for a notification rule """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/notificationRules/{ruleID}/labels', tags=['label_operations', 'notification_rule_handling'], ) def post_notification_rule_i_d_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), rule_i_d: str = Path(..., alias='ruleID'), body: LabelMapping = ..., ): """ Add a label to a notification rule """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/notificationRules/{ruleID}/labels/{labelID}', tags=['label_operations', 'notification_rule_handling'], ) def delete_notification_rules_i_d_labels_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), rule_i_d: str = Path(..., alias='ruleID'), label_i_d: str = Path(..., alias='labelID'), ): """ Delete label from a notification rule """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/notificationRules/{ruleID}/query', tags=['notification_rule_handling']) def get_notification_rules_i_d_query( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), rule_i_d: str = Path(..., alias='ruleID'), ): """ Retrieve a notification rule query """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/orgs', tags=['organization_operations']) def get_orgs( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), offset: Optional[conint(ge=0)] = None, limit: Optional[conint(ge=1, le=100)] = 20, descending: Optional[bool] = False, org: Optional[str] = None, org_i_d: Optional[str] = Query(None, alias='orgID'), user_i_d: Optional[str] = Query(None, alias='userID'), ): """ List all organizations """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/orgs', tags=['organization_operations']) def post_orgs( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), body: PostOrganizationRequest = ..., ): """ Create an organization """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete('/orgs/{orgID}', tags=['organization_operations']) def delete_orgs_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), org_i_d: str = Path(..., alias='orgID'), ): """ Delete an organization """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/orgs/{orgID}', tags=['organization_operations']) def get_orgs_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), org_i_d: str = Path(..., alias='orgID'), ): """ Retrieve an organization """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.patch('/orgs/{orgID}', tags=['organization_operations']) def patch_orgs_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), org_i_d: str = Path(..., alias='orgID'), body: PatchOrganizationRequest = ..., ): """ Update an organization """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/orgs/{orgID}/members', tags=['organization_operations', 'user_account_management'] ) def get_orgs_i_d_members( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), org_i_d: str = Path(..., alias='orgID'), ): """ List all members of an organization """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/orgs/{orgID}/members', tags=['organization_operations', 'user_account_management'] ) def post_orgs_i_d_members( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), org_i_d: str = Path(..., alias='orgID'), body: AddResourceMemberRequestBody = ..., ): """ Add a member to an organization """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/orgs/{orgID}/members/{userID}', tags=['user_account_management', 'organization_operations'], ) def delete_orgs_i_d_members_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), user_i_d: str = Path(..., alias='userID'), org_i_d: str = Path(..., alias='orgID'), ): """ Remove a member from an organization """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/orgs/{orgID}/owners', tags=['organization_operations', 'user_account_management'] ) def get_orgs_i_d_owners( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), org_i_d: str = Path(..., alias='orgID'), ): """ List all owners of an organization """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/orgs/{orgID}/owners', tags=['organization_operations', 'user_account_management'] ) def post_orgs_i_d_owners( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), org_i_d: str = Path(..., alias='orgID'), body: AddResourceMemberRequestBody = ..., ): """ Add an owner to an organization """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/orgs/{orgID}/owners/{userID}', tags=['user_account_management', 'organization_operations'], ) def delete_orgs_i_d_owners_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), user_i_d: str = Path(..., alias='userID'), org_i_d: str = Path(..., alias='orgID'), ): """ Remove an owner from an organization """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/orgs/{orgID}/secrets', tags=['organization_operations']) def get_orgs_i_d_secrets( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), org_i_d: str = Path(..., alias='orgID'), ): """ List all secret keys for an organization """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.patch('/orgs/{orgID}/secrets', tags=['organization_operations']) def patch_orgs_i_d_secrets( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), org_i_d: str = Path(..., alias='orgID'), body: Secrets = ..., ): """ Update secrets in an organization """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/orgs/{orgID}/secrets/delete', tags=['organization_operations']) def post_orgs_i_d_secrets( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), org_i_d: str = Path(..., alias='orgID'), body: SecretKeys = ..., ): """ Delete secrets from an organization """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/query', tags=['influxdb_query_handling']) def post_query( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), accept__encoding: Optional[AcceptEncoding] = Header( 'identity', alias='Accept-Encoding' ), content__type: Optional[ContentType] = Header(None, alias='Content-Type'), org: Optional[str] = None, org_i_d: Optional[str] = Query(None, alias='orgID'), body: QueryPostRequest = None, ): """ Query InfluxDB """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/query/analyze', tags=['influxdb_query_handling']) def post_query_analyze( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), content__type: Optional[ContentType1] = Header(None, alias='Content-Type'), body: Query = None, ): """ Analyze an InfluxQL or Flux query """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/query/ast', description=""" Analyzes flux query and generates a query specification. """, tags=['influxdb_query_handling'], ) def post_query_ast( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), content__type: Optional[ContentType1] = Header(None, alias='Content-Type'), body: LanguageRequest = None, ): """ Generate an Abstract Syntax Tree (AST) from a query """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/query/suggestions', tags=['influxdb_query_handling']) def get_query_suggestions( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span') ): """ Retrieve query suggestions """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/query/suggestions/{name}', tags=['influxdb_query_handling']) def get_query_suggestions_name( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), name: str = ..., ): """ Retrieve query suggestions for a branching suggestion """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/ready', tags=['system_health_check']) def get_ready(zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span')): """ Get the readiness of an instance at startup """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/scrapers', tags=['scraper_target_management', 'scraper_member_handling']) def get_scrapers( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), name: Optional[str] = None, id: Optional[Id] = None, org_i_d: Optional[str] = Query(None, alias='orgID'), org: Optional[str] = None, ): """ List all scraper targets """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/scrapers', tags=['scraper_target_management']) def post_scrapers( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), body: ScraperTargetRequest = ..., ): """ Create a scraper target """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete('/scrapers/{scraperTargetID}', tags=['scraper_target_management']) def delete_scrapers_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), scraper_target_i_d: str = Path(..., alias='scraperTargetID'), ): """ Delete a scraper target """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/scrapers/{scraperTargetID}', tags=['scraper_target_management']) def get_scrapers_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), scraper_target_i_d: str = Path(..., alias='scraperTargetID'), ): """ Retrieve a scraper target """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.patch('/scrapers/{scraperTargetID}', tags=['scraper_target_management']) def patch_scrapers_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), scraper_target_i_d: str = Path(..., alias='scraperTargetID'), body: ScraperTargetRequest = ..., ): """ Update a scraper target """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/scrapers/{scraperTargetID}/labels', tags=['label_operations', 'scraper_target_management'], ) def get_scrapers_i_d_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), scraper_target_i_d: str = Path(..., alias='scraperTargetID'), ): """ List all labels for a scraper target """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/scrapers/{scraperTargetID}/labels', tags=['scraper_target_management', 'label_operations'], ) def post_scrapers_i_d_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), scraper_target_i_d: str = Path(..., alias='scraperTargetID'), body: LabelMapping = ..., ): """ Add a label to a scraper target """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/scrapers/{scraperTargetID}/labels/{labelID}', tags=['label_operations', 'scraper_target_management'], ) def delete_scrapers_i_d_labels_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), scraper_target_i_d: str = Path(..., alias='scraperTargetID'), label_i_d: str = Path(..., alias='labelID'), ): """ Delete a label from a scraper target """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/scrapers/{scraperTargetID}/members', tags=['scraper_member_handling', 'user_account_management'], ) def get_scrapers_i_d_members( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), scraper_target_i_d: str = Path(..., alias='scraperTargetID'), ): """ List all users with member privileges for a scraper target """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/scrapers/{scraperTargetID}/members', tags=['scraper_member_handling', 'scraper_target_management'], ) def post_scrapers_i_d_members( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), scraper_target_i_d: str = Path(..., alias='scraperTargetID'), body: AddResourceMemberRequestBody = ..., ): """ Add a member to a scraper target """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/scrapers/{scraperTargetID}/members/{userID}', tags=['scraper_target_management', 'scraper_member_handling'], ) def delete_scrapers_i_d_members_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), user_i_d: str = Path(..., alias='userID'), scraper_target_i_d: str = Path(..., alias='scraperTargetID'), ): """ Remove a member from a scraper target """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/scrapers/{scraperTargetID}/owners', tags=['scraper_member_handling', 'scraper_target_management'], ) def get_scrapers_i_d_owners( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), scraper_target_i_d: str = Path(..., alias='scraperTargetID'), ): """ List all owners of a scraper target """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/scrapers/{scraperTargetID}/owners', tags=['scraper_target_management', 'scraper_member_handling'], ) def post_scrapers_i_d_owners( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), scraper_target_i_d: str = Path(..., alias='scraperTargetID'), body: AddResourceMemberRequestBody = ..., ): """ Add an owner to a scraper target """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/scrapers/{scraperTargetID}/owners/{userID}', tags=['scraper_member_handling', 'scraper_target_management'], ) def delete_scrapers_i_d_owners_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), user_i_d: str = Path(..., alias='userID'), scraper_target_i_d: str = Path(..., alias='scraperTargetID'), ): """ Remove an owner from a scraper target """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/setup', description=""" Returns `true` if no default user, organization, or bucket has been created. """, tags=[ 'organization_operations', 'user_account_management', 'bucket_operations', 'check_operations', ], ) def get_setup(zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span')): """ Check if database has default user, org, bucket """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/setup', description=""" Post an onboarding request to set up initial user, org and bucket. """, tags=['user_account_management', 'organization_operations', 'bucket_operations'], ) def post_setup( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), body: OnboardingRequest = ..., ): """ Set up initial user, org and bucket """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/signin', tags=['authorization_handling', 'user_session_operations'], security=[ HTTPBasic(name="None"), ], ) def post_signin(zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span')): """ Exchange basic auth credentials for session """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/signout', tags=['user_session_operations']) def post_signout( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span') ): """ Expire the current session """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/sources', tags=['source_operations']) def get_sources( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), org: Optional[str] = None, ): """ List all sources """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/sources', tags=['source_operations']) def post_sources( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), body: Source = ..., ): """ Create a source """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete('/sources/{sourceID}', tags=['source_operations']) def delete_sources_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), source_i_d: str = Path(..., alias='sourceID'), ): """ Delete a source """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/sources/{sourceID}', tags=['source_operations']) def get_sources_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), source_i_d: str = Path(..., alias='sourceID'), ): """ Retrieve a source """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.patch('/sources/{sourceID}', tags=['source_operations']) def patch_sources_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), source_i_d: str = Path(..., alias='sourceID'), body: Source = ..., ): """ Update a Source """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/sources/{sourceID}/buckets', tags=['bucket_operations', 'source_operations']) def get_sources_i_d_buckets( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), source_i_d: str = Path(..., alias='sourceID'), org: Optional[str] = None, ): """ Get buckets in a source """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/sources/{sourceID}/health', tags=['system_health_check', 'source_operations'] ) def get_sources_i_d_health( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), source_i_d: str = Path(..., alias='sourceID'), ): """ Get the health of a source """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/stacks', tags=['document_template_operations', 'template_operations']) def list_stacks( org_i_d: str = Query(..., alias='orgID'), name: Optional[str] = None, stack_i_d: Optional[str] = Query(None, alias='stackID'), ): """ List all installed InfluxDB templates """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/stacks', tags=['stack_lifecycle_management']) def create_stack(body: StacksPostRequest): """ Create a new stack """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/stacks/{stack_id}', tags=['stack_lifecycle_management', 'organization_operations'] ) def delete_stack(stack_id: str, org_i_d: str = Query(..., alias='orgID')): """ Delete a stack and associated resources """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/stacks/{stack_id}', tags=['stack_lifecycle_management']) def read_stack(stack_id: str): """ Retrieve a stack """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.patch('/stacks/{stack_id}', tags=['stack_lifecycle_management']) def update_stack(stack_id: str, body: StacksStackIdPatchRequest = ...): """ Update an InfluxDB Stack """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/stacks/{stack_id}/uninstall', tags=['stack_lifecycle_management']) def uninstall_stack(stack_id: str): """ Uninstall an InfluxDB Stack """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/tasks', tags=['task_operations']) def get_tasks( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), name: Optional[str] = None, after: Optional[str] = None, user: Optional[str] = None, org: Optional[str] = None, org_i_d: Optional[str] = Query(None, alias='orgID'), status: Optional[Status9] = None, limit: Optional[conint(ge=1, le=500)] = 100, ): """ List all tasks """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/tasks', tags=['task_operations']) def post_tasks( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), body: TaskCreateRequest = ..., ): """ Create a new task """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/tasks/{taskID}', description=""" Deletes a task and all associated records """, tags=['task_operations'], ) def delete_tasks_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), task_i_d: str = Path(..., alias='taskID'), ): """ Delete a task """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/tasks/{taskID}', tags=['task_operations']) def get_tasks_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), task_i_d: str = Path(..., alias='taskID'), ): """ Retrieve a task """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.patch( '/tasks/{taskID}', description=""" Update a task. This will cancel all queued runs. """, tags=['task_operations'], ) def patch_tasks_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), task_i_d: str = Path(..., alias='taskID'), body: TaskUpdateRequest = ..., ): """ Update a task """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/tasks/{taskID}/labels', tags=['label_operations', 'task_operations']) def get_tasks_i_d_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), task_i_d: str = Path(..., alias='taskID'), ): """ List all labels for a task """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/tasks/{taskID}/labels', tags=['label_operations', 'task_operations']) def post_tasks_i_d_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), task_i_d: str = Path(..., alias='taskID'), body: LabelMapping = ..., ): """ Add a label to a task """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/tasks/{taskID}/labels/{labelID}', tags=['label_operations', 'task_operations'] ) def delete_tasks_i_d_labels_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), task_i_d: str = Path(..., alias='taskID'), label_i_d: str = Path(..., alias='labelID'), ): """ Delete a label from a task """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/tasks/{taskID}/logs', tags=['task_operations']) def get_tasks_i_d_logs( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), task_i_d: str = Path(..., alias='taskID'), ): """ Retrieve all logs for a task """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/tasks/{taskID}/members', tags=['task_operations', 'organization_operations']) def get_tasks_i_d_members( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), task_i_d: str = Path(..., alias='taskID'), ): """ List all task members """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/tasks/{taskID}/members', tags=['task_operations']) def post_tasks_i_d_members( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), task_i_d: str = Path(..., alias='taskID'), body: AddResourceMemberRequestBody = ..., ): """ Add a member to a task """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/tasks/{taskID}/members/{userID}', tags=['task_operations', 'user_account_management'], ) def delete_tasks_i_d_members_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), user_i_d: str = Path(..., alias='userID'), task_i_d: str = Path(..., alias='taskID'), ): """ Remove a member from a task """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/tasks/{taskID}/owners', tags=['task_operations']) def get_tasks_i_d_owners( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), task_i_d: str = Path(..., alias='taskID'), ): """ List all owners of a task """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/tasks/{taskID}/owners', tags=['task_operations']) def post_tasks_i_d_owners( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), task_i_d: str = Path(..., alias='taskID'), body: AddResourceMemberRequestBody = ..., ): """ Add an owner to a task """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/tasks/{taskID}/owners/{userID}', tags=['task_operations', 'user_account_management'], ) def delete_tasks_i_d_owners_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), user_i_d: str = Path(..., alias='userID'), task_i_d: str = Path(..., alias='taskID'), ): """ Remove an owner from a task """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/tasks/{taskID}/runs', tags=['task_operations']) def get_tasks_i_d_runs( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), task_i_d: str = Path(..., alias='taskID'), after: Optional[str] = None, limit: Optional[conint(ge=1, le=500)] = 100, after_time: Optional[datetime] = Query(None, alias='afterTime'), before_time: Optional[datetime] = Query(None, alias='beforeTime'), ): """ List runs for a task """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/tasks/{taskID}/runs', tags=['task_operations']) def post_tasks_i_d_runs( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), task_i_d: str = Path(..., alias='taskID'), body: RunManually = None, ): """ Manually start a task run, overriding the current schedule """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete('/tasks/{taskID}/runs/{runID}', tags=['task_operations']) def delete_tasks_i_d_runs_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), task_i_d: str = Path(..., alias='taskID'), run_i_d: str = Path(..., alias='runID'), ): """ Cancel a running task """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/tasks/{taskID}/runs/{runID}', tags=['task_operations']) def get_tasks_i_d_runs_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), task_i_d: str = Path(..., alias='taskID'), run_i_d: str = Path(..., alias='runID'), ): """ Retrieve a single run for a task """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/tasks/{taskID}/runs/{runID}/logs', tags=['task_operations']) def get_tasks_i_d_runs_i_d_logs( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), task_i_d: str = Path(..., alias='taskID'), run_i_d: str = Path(..., alias='runID'), ): """ Retrieve all logs for a run """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/tasks/{taskID}/runs/{runID}/retry', tags=['task_operations']) def post_tasks_i_d_runs_i_d_retry( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), task_i_d: str = Path(..., alias='taskID'), run_i_d: str = Path(..., alias='runID'), ): """ Retry a task run """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/telegraf/plugins', tags=['telegraf_configuration_management']) def get_telegraf_plugins( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), type: Optional[str] = None, ): """ List all Telegraf plugins """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/telegrafs', tags=['telegraf_configuration_management']) def get_telegrafs( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), org_i_d: Optional[str] = Query(None, alias='orgID'), ): """ List all Telegraf configurations """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/telegrafs', tags=['telegraf_configuration_management']) def post_telegrafs( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), body: TelegrafRequest = ..., ): """ Create a Telegraf configuration """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete('/telegrafs/{telegrafID}', tags=['telegraf_configuration_management']) def delete_telegrafs_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), telegraf_i_d: str = Path(..., alias='telegrafID'), ): """ Delete a Telegraf configuration """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/telegrafs/{telegrafID}', tags=['telegraf_configuration_management']) def get_telegrafs_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), telegraf_i_d: str = Path(..., alias='telegrafID'), accept: Optional[Accept] = Header('application/toml', alias='Accept'), ): """ Retrieve a Telegraf configuration """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.put('/telegrafs/{telegrafID}', tags=['telegraf_configuration_management']) def put_telegrafs_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), telegraf_i_d: str = Path(..., alias='telegrafID'), body: TelegrafRequest = ..., ): """ Update a Telegraf configuration """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/telegrafs/{telegrafID}/labels', tags=['telegraf_configuration_management', 'label_operations'], ) def get_telegrafs_i_d_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), telegraf_i_d: str = Path(..., alias='telegrafID'), ): """ List all labels for a Telegraf config """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/telegrafs/{telegrafID}/labels', tags=['label_operations', 'telegraf_configuration_management'], ) def post_telegrafs_i_d_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), telegraf_i_d: str = Path(..., alias='telegrafID'), body: LabelMapping = ..., ): """ Add a label to a Telegraf config """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/telegrafs/{telegrafID}/labels/{labelID}', tags=['label_operations', 'telegraf_configuration_management'], ) def delete_telegrafs_i_d_labels_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), telegraf_i_d: str = Path(..., alias='telegrafID'), label_i_d: str = Path(..., alias='labelID'), ): """ Delete a label from a Telegraf config """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/telegrafs/{telegrafID}/members', tags=['telegraf_configuration_management', 'user_account_management'], ) def get_telegrafs_i_d_members( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), telegraf_i_d: str = Path(..., alias='telegrafID'), ): """ List all users with member privileges for a Telegraf config """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/telegrafs/{telegrafID}/members', tags=['telegraf_configuration_management']) def post_telegrafs_i_d_members( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), telegraf_i_d: str = Path(..., alias='telegrafID'), body: AddResourceMemberRequestBody = ..., ): """ Add a member to a Telegraf config """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/telegrafs/{telegrafID}/members/{userID}', tags=['telegraf_configuration_management', 'scraper_member_handling'], ) def delete_telegrafs_i_d_members_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), user_i_d: str = Path(..., alias='userID'), telegraf_i_d: str = Path(..., alias='telegrafID'), ): """ Remove a member from a Telegraf config """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/telegrafs/{telegrafID}/owners', tags=['telegraf_configuration_management']) def get_telegrafs_i_d_owners( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), telegraf_i_d: str = Path(..., alias='telegrafID'), ): """ List all owners of a Telegraf configuration """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/telegrafs/{telegrafID}/owners', tags=['telegraf_configuration_management']) def post_telegrafs_i_d_owners( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), telegraf_i_d: str = Path(..., alias='telegrafID'), body: AddResourceMemberRequestBody = ..., ): """ Add an owner to a Telegraf configuration """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/telegrafs/{telegrafID}/owners/{userID}', tags=['telegraf_configuration_management', 'user_account_management'], ) def delete_telegrafs_i_d_owners_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), user_i_d: str = Path(..., alias='userID'), telegraf_i_d: str = Path(..., alias='telegrafID'), ): """ Remove an owner from a Telegraf config """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/templates/apply', tags=['template_application_handling', 'template_operations'] ) def apply_template(body: TemplateApply): """ Apply or dry-run an InfluxDB Template """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/templates/export', tags=[ 'template_exporting_operations', 'document_template_operations', 'template_operations', ], ) def export_template(body: TemplatesExportPostRequest = None): """ Export a new Influx Template """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/users', tags=['user_account_management']) def get_users( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), offset: Optional[conint(ge=0)] = None, limit: Optional[conint(ge=1, le=100)] = 20, after: Optional[str] = None, name: Optional[str] = None, id: Optional[str] = None, ): """ List all users """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/users', tags=['user_account_management']) def post_users( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), body: User = ..., ): """ Create a user """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete('/users/{userID}', tags=['user_account_management']) def delete_users_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), user_i_d: str = Path(..., alias='userID'), ): """ Delete a user """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/users/{userID}', tags=['user_account_management']) def get_users_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), user_i_d: str = Path(..., alias='userID'), ): """ Retrieve a user """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.patch('/users/{userID}', tags=['user_account_management']) def patch_users_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), user_i_d: str = Path(..., alias='userID'), body: User = ..., ): """ Update a user """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/users/{userID}/password', tags=['user_account_management'], security=[ HTTPBasic(name="None"), ], ) def post_users_i_d_password( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), user_i_d: str = Path(..., alias='userID'), body: PasswordResetBody = ..., ): """ Update a password """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/variables', tags=['application_variable_management']) def get_variables( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), org: Optional[str] = None, org_i_d: Optional[str] = Query(None, alias='orgID'), ): """ List all variables """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/variables', tags=['application_variable_management']) def post_variables( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), body: Variable2 = ..., ): """ Create a variable """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete('/variables/{variableID}', tags=['application_variable_management']) def delete_variables_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), variable_i_d: str = Path(..., alias='variableID'), ): """ Delete a variable """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get('/variables/{variableID}', tags=['application_variable_management']) def get_variables_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), variable_i_d: str = Path(..., alias='variableID'), ): """ Retrieve a variable """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.patch('/variables/{variableID}', tags=['application_variable_management']) def patch_variables_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), variable_i_d: str = Path(..., alias='variableID'), body: Variable2 = ..., ): """ Update a variable """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.put('/variables/{variableID}', tags=['application_variable_management']) def put_variables_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), variable_i_d: str = Path(..., alias='variableID'), body: Variable2 = ..., ): """ Replace a variable """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/variables/{variableID}/labels', tags=['label_operations', 'application_variable_management'], ) def get_variables_i_d_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), variable_i_d: str = Path(..., alias='variableID'), ): """ List all labels for a variable """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/variables/{variableID}/labels', tags=['label_operations', 'application_variable_management'], ) def post_variables_i_d_labels( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), variable_i_d: str = Path(..., alias='variableID'), body: LabelMapping = ..., ): """ Add a label to a variable """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/variables/{variableID}/labels/{labelID}', tags=['label_operations', 'application_variable_management'], ) def delete_variables_i_d_labels_i_d( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), variable_i_d: str = Path(..., alias='variableID'), label_i_d: str = Path(..., alias='labelID'), ): """ Delete a label from a variable """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post('/write', tags=['data_writing_operations']) def post_write( zap__trace__span: Optional[str] = Header(None, alias='Zap-Trace-Span'), content__encoding: Optional[ContentEncoding] = Header( 'identity', alias='Content-Encoding' ), content__type: Optional[ContentType3] = Header( 'text/plain; charset=utf-8', alias='Content-Type' ), content__length: Optional[int] = Header(None, alias='Content-Length'), accept: Optional[Accept1] = Header('application/json', alias='Accept'), org: str = ..., org_i_d: Optional[str] = Query(None, alias='orgID'), bucket: str = ..., precision: Optional[WritePrecision] = None, ): """ Write time series data into InfluxDB """ 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)