main.py•31.5 kB
# generated by fastapi-codegen:
# filename: openapi.yaml
# timestamp: 2025-06-29T05:09:48+00:00
import argparse
import json
import os
from typing import *
from typing import Optional, Union
from uuid import UUID
from autogen.mcp.mcp_proxy import MCPProxy
from autogen.mcp.mcp_proxy.security import APIKeyHeader, BaseSecurity
from fastapi import Path, Query, UploadFile
from pydantic import conint
from starlette.requests import Request
from models import (
ClassificationType,
CustomVisionError,
Domain,
DomainsGetResponse,
Export,
Flavor,
Ids,
ImageCreateSummary,
ImageFileCreateBatch,
ImageIdCreateBatch,
ImageIds,
ImageIds2,
ImagePrediction,
ImageRegionCreateBatch,
ImageRegionCreateSummary,
ImageRegionProposal,
ImageTagCreateBatch,
ImageTagCreateSummary,
ImageUrl,
ImageUrlCreateBatch,
Iteration,
IterationPerformance,
OrderBy1,
Platform,
PredictionQueryResult,
PredictionQueryToken,
Project,
ProjectExport,
ProjectsGetResponse,
ProjectsProjectIdImagesIdGetResponse,
ProjectsProjectIdImagesSuggestedCountPostResponse,
ProjectsProjectIdImagesTaggedGetResponse,
ProjectsProjectIdImagesUntaggedGetResponse,
ProjectsProjectIdIterationsGetResponse,
ProjectsProjectIdIterationsIterationIdExportGetResponse,
ProjectsProjectIdIterationsIterationIdPerformanceImagesGetResponse,
ProjectsProjectIdTagsandregionsSuggestionsPostResponse,
ProjectsProjectIdTagsGetResponse,
RegionIds,
SuggestedTagAndRegionQuery,
SuggestedTagAndRegionQueryToken,
Tag,
TagFilter,
TagIds,
TagIds2,
TagIds3,
TagIds5,
TargetExportPlatforms,
TrainingParameters,
TrainingType,
Type1,
)
app = MCPProxy(
contact={'x-twitter': 'microsoft'},
title='Custom Vision Training Client',
version='3.2',
servers=[
{
'url': 'https://southcentralus.api.cognitive.microsoft.com/customvision/v3.2/training'
},
{
'url': '{Endpoint}/customvision/v3.2/training',
'variables': {
'Endpoint': {
'default': 'none',
'description': 'Supported Cognitive Services endpoints.',
'x-ms-parameter-location': 'client',
'x-ms-skip-url-encoding': True,
}
},
},
],
)
@app.get(
'/domains',
tags=['domain_administration'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def get_domains():
"""
Get a list of the available domains.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/domains/{domainId}',
tags=['domain_administration'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def get_domain(domain_id: UUID = Path(..., alias='domainId')):
"""
Get information about a specific domain.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/projects',
tags=['project_lifecycle_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def get_projects():
"""
Get your projects.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/projects',
tags=['project_lifecycle_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def create_project(
name: str,
description: Optional[str] = None,
domain_id: Optional[UUID] = Query(None, alias='domainId'),
classification_type: Optional[ClassificationType] = Query(
None, alias='classificationType'
),
target_export_platforms: Optional[TargetExportPlatforms] = Query(
None, alias='targetExportPlatforms'
),
):
"""
Create a project.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/projects/import',
tags=['project_lifecycle_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def import_project(token: str):
"""
Imports a project.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.delete(
'/projects/{projectId}',
tags=['project_lifecycle_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def delete_project(project_id: UUID = Path(..., alias='projectId')):
"""
Delete a specific project.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/projects/{projectId}',
tags=['project_lifecycle_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def get_project(project_id: UUID = Path(..., alias='projectId')):
"""
Get a specific project.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.patch(
'/projects/{projectId}',
tags=['project_lifecycle_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def update_project(
project_id: UUID = Path(..., alias='projectId'), body: Project = ...
):
"""
Update a specific project.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/projects/{projectId}/export',
tags=['project_lifecycle_management', 'export_management_operations'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def export_project(project_id: UUID = Path(..., alias='projectId')):
"""
Exports a project.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.delete(
'/projects/{projectId}/images',
tags=[
'project_lifecycle_management',
'image_handling_operations',
'iteration_control',
'training_process_management',
],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def delete_images(
project_id: UUID = Path(..., alias='projectId'),
image_ids: Optional[ImageIds] = Query(None, alias='imageIds'),
all_images: Optional[bool] = Query(None, alias='allImages'),
all_iterations: Optional[bool] = Query(None, alias='allIterations'),
):
"""
Delete images from the set of training images.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/projects/{projectId}/images',
description=""" This API accepts body content as multipart/form-data and application/octet-stream. When using multipart
multiple image files can be sent at once, with a maximum of 64 files """,
tags=[
'project_lifecycle_management',
'image_handling_operations',
'training_process_management',
],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def create_images_from_data(
project_id: UUID = Path(..., alias='projectId'),
tag_ids: Optional[TagIds] = Query(None, alias='tagIds'),
file: UploadFile = ...,
):
"""
Add the provided images to the set of training images.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/projects/{projectId}/images/files',
description=""" This API accepts a batch of files, and optionally tags, to create images. There is a limit of 64 images and 20 tags. """,
tags=[
'project_lifecycle_management',
'image_handling_operations',
'training_process_management',
],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def create_images_from_files(
project_id: UUID = Path(..., alias='projectId'), body: ImageFileCreateBatch = ...
):
"""
Add the provided batch of images to the set of training images.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/projects/{projectId}/images/id',
description=""" This API will return a set of Images for the specified tags and optionally iteration. If no iteration is specified the
current workspace is used. """,
tags=[
'project_lifecycle_management',
'image_handling_operations',
'iteration_control',
],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def get_images_by_ids(
project_id: UUID = Path(..., alias='projectId'),
image_ids: Optional[ImageIds] = Query(None, alias='imageIds'),
iteration_id: Optional[UUID] = Query(None, alias='iterationId'),
):
"""
Get images by id for a given project iteration.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/projects/{projectId}/images/predictions',
description=""" This API creates a batch of images from predicted images specified. There is a limit of 64 images and 20 tags. """,
tags=[
'project_lifecycle_management',
'image_handling_operations',
'prediction_management',
'training_process_management',
],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def create_images_from_predictions(
project_id: UUID = Path(..., alias='projectId'), body: ImageIdCreateBatch = ...
):
"""
Add the specified predicted images to the set of training images.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.delete(
'/projects/{projectId}/images/regions',
tags=[
'project_lifecycle_management',
'image_handling_operations',
'image_region_operations',
],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def delete_image_regions(
project_id: UUID = Path(..., alias='projectId'),
region_ids: RegionIds = Query(..., alias='regionIds'),
):
"""
Delete a set of image regions.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/projects/{projectId}/images/regions',
description=""" This API accepts a batch of image regions, and optionally tags, to update existing images with region information.
There is a limit of 64 entries in the batch. """,
tags=[
'image_region_operations',
'image_handling_operations',
'project_lifecycle_management',
],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def create_image_regions(
project_id: UUID = Path(..., alias='projectId'), body: ImageRegionCreateBatch = ...
):
"""
Create a set of image regions.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/projects/{projectId}/images/suggested',
description=""" This API will fetch untagged images filtered by suggested tags Ids. It returns an empty array if no images are found. """,
tags=[
'image_handling_operations',
'tagging_system_management',
'project_lifecycle_management',
],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def query_suggested_images(
project_id: UUID = Path(..., alias='projectId'),
iteration_id: UUID = Query(..., alias='iterationId'),
body: SuggestedTagAndRegionQueryToken = ...,
):
"""
Get untagged images whose suggested tags match given tags. Returns empty array if no images are found.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/projects/{projectId}/images/suggested/count',
description=""" This API takes in tagIds to get count of untagged images per suggested tags for a given threshold. """,
tags=[
'image_handling_operations',
'tagging_system_management',
'prediction_management',
],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def query_suggested_image_count(
project_id: UUID = Path(..., alias='projectId'),
iteration_id: UUID = Query(..., alias='iterationId'),
body: TagFilter = ...,
):
"""
Get count of images whose suggested tags match given tags and their probabilities are greater than or equal to the given threshold. Returns count as 0 if none found.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/projects/{projectId}/images/tagged',
description=""" This API supports batching and range selection. By default it will only return first 50 images matching images.
Use the {take} and {skip} parameters to control how many images to return in a given batch.
The filtering is on an and/or relationship. For example, if the provided tag ids are for the "Dog" and
"Cat" tags, then only images tagged with Dog and/or Cat will be returned """,
tags=[
'project_lifecycle_management',
'image_handling_operations',
'tagging_system_management',
'iteration_control',
],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def get_tagged_images(
project_id: UUID = Path(..., alias='projectId'),
iteration_id: Optional[UUID] = Query(None, alias='iterationId'),
tag_ids: Optional[TagIds] = Query(None, alias='tagIds'),
order_by: Optional[OrderBy1] = Query(None, alias='orderBy'),
take: Optional[conint(ge=0, le=256)] = 50,
skip: Optional[int] = 0,
):
"""
Get tagged images for a given project iteration.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/projects/{projectId}/images/tagged/count',
description=""" The filtering is on an and/or relationship. For example, if the provided tag ids are for the "Dog" and
"Cat" tags, then only images tagged with Dog and/or Cat will be returned """,
tags=['image_handling_operations', 'tagging_system_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def get_tagged_image_count(
project_id: UUID = Path(..., alias='projectId'),
iteration_id: Optional[UUID] = Query(None, alias='iterationId'),
tag_ids: Optional[TagIds2] = Query(None, alias='tagIds'),
):
"""
Gets the number of images tagged with the provided {tagIds}.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.delete(
'/projects/{projectId}/images/tags',
tags=['image_handling_operations', 'tagging_system_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def delete_image_tags(
project_id: UUID = Path(..., alias='projectId'),
image_ids: ImageIds2 = Query(..., alias='imageIds'),
tag_ids: TagIds3 = Query(..., alias='tagIds'),
):
"""
Remove a set of tags from a set of images.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/projects/{projectId}/images/tags',
tags=['image_handling_operations', 'tagging_system_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def create_image_tags(
project_id: UUID = Path(..., alias='projectId'), body: ImageTagCreateBatch = ...
):
"""
Associate a set of images with a set of tags.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/projects/{projectId}/images/untagged',
description=""" This API supports batching and range selection. By default it will only return first 50 images matching images.
Use the {take} and {skip} parameters to control how many images to return in a given batch. """,
tags=[
'project_lifecycle_management',
'image_handling_operations',
'iteration_control',
],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def get_untagged_images(
project_id: UUID = Path(..., alias='projectId'),
iteration_id: Optional[UUID] = Query(None, alias='iterationId'),
order_by: Optional[OrderBy1] = Query(None, alias='orderBy'),
take: Optional[conint(ge=0, le=256)] = 50,
skip: Optional[int] = 0,
):
"""
Get untagged images for a given project iteration.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/projects/{projectId}/images/untagged/count',
description=""" This API returns the images which have no tags for a given project and optionally an iteration. If no iteration is specified the
current workspace is used. """,
tags=['image_handling_operations', 'project_lifecycle_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def get_untagged_image_count(
project_id: UUID = Path(..., alias='projectId'),
iteration_id: Optional[UUID] = Query(None, alias='iterationId'),
):
"""
Gets the number of untagged images.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/projects/{projectId}/images/urls',
description=""" This API accepts a batch of urls, and optionally tags, to create images. There is a limit of 64 images and 20 tags. """,
tags=[
'project_lifecycle_management',
'image_handling_operations',
'training_process_management',
],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def create_images_from_urls(
project_id: UUID = Path(..., alias='projectId'), body: ImageUrlCreateBatch = ...
):
"""
Add the provided images urls to the set of training images.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/projects/{projectId}/images/{imageId}/regionproposals',
description=""" This API will get region proposals for an image along with confidences for the region. It returns an empty array if no proposals are found. """,
tags=[
'image_region_operations',
'image_handling_operations',
'project_lifecycle_management',
],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def get_image_region_proposals(
project_id: UUID = Path(..., alias='projectId'),
image_id: UUID = Path(..., alias='imageId'),
):
"""
Get region proposals for an image. Returns empty array if no proposals are found.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/projects/{projectId}/iterations',
tags=['iteration_control', 'project_lifecycle_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def get_iterations(project_id: UUID = Path(..., alias='projectId')):
"""
Get iterations for the project.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.delete(
'/projects/{projectId}/iterations/{iterationId}',
tags=['project_lifecycle_management', 'iteration_control'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def delete_iteration(
project_id: UUID = Path(..., alias='projectId'),
iteration_id: UUID = Path(..., alias='iterationId'),
):
"""
Delete a specific iteration of a project.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/projects/{projectId}/iterations/{iterationId}',
tags=['iteration_control', 'project_lifecycle_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def get_iteration(
project_id: UUID = Path(..., alias='projectId'),
iteration_id: UUID = Path(..., alias='iterationId'),
):
"""
Get a specific iteration.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.patch(
'/projects/{projectId}/iterations/{iterationId}',
tags=['project_lifecycle_management', 'iteration_control'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def update_iteration(
project_id: UUID = Path(..., alias='projectId'),
iteration_id: UUID = Path(..., alias='iterationId'),
body: Iteration = ...,
):
"""
Update a specific iteration.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/projects/{projectId}/iterations/{iterationId}/export',
tags=['export_management_operations', 'iteration_control'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def get_exports(
project_id: UUID = Path(..., alias='projectId'),
iteration_id: UUID = Path(..., alias='iterationId'),
):
"""
Get the list of exports for a specific iteration.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/projects/{projectId}/iterations/{iterationId}/export',
tags=['export_management_operations', 'iteration_control'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def export_iteration(
project_id: UUID = Path(..., alias='projectId'),
iteration_id: UUID = Path(..., alias='iterationId'),
platform: Platform = ...,
flavor: Optional[Flavor] = None,
):
"""
Export a trained iteration.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/projects/{projectId}/iterations/{iterationId}/performance',
tags=['iteration_control', 'performance_analysis'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def get_iteration_performance(
project_id: UUID = Path(..., alias='projectId'),
iteration_id: UUID = Path(..., alias='iterationId'),
threshold: Optional[float] = None,
overlap_threshold: Optional[float] = Query(None, alias='overlapThreshold'),
):
"""
Get detailed performance information about an iteration.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/projects/{projectId}/iterations/{iterationId}/performance/images',
description=""" This API supports batching and range selection. By default it will only return first 50 images matching images.
Use the {take} and {skip} parameters to control how many images to return in a given batch.
The filtering is on an and/or relationship. For example, if the provided tag ids are for the "Dog" and
"Cat" tags, then only images tagged with Dog and/or Cat will be returned """,
tags=[
'project_lifecycle_management',
'image_handling_operations',
'prediction_management',
],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def get_image_performances(
project_id: UUID = Path(..., alias='projectId'),
iteration_id: UUID = Path(..., alias='iterationId'),
tag_ids: Optional[TagIds3] = Query(None, alias='tagIds'),
order_by: Optional[OrderBy1] = Query(None, alias='orderBy'),
take: Optional[conint(ge=0, le=256)] = 50,
skip: Optional[int] = 0,
):
"""
Get image with its prediction for a given project iteration.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/projects/{projectId}/iterations/{iterationId}/performance/images/count',
description=""" The filtering is on an and/or relationship. For example, if the provided tag ids are for the "Dog" and
"Cat" tags, then only images tagged with Dog and/or Cat will be returned """,
tags=[
'image_handling_operations',
'iteration_control',
'prediction_management',
'performance_analysis',
],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def get_image_performance_count(
project_id: UUID = Path(..., alias='projectId'),
iteration_id: UUID = Path(..., alias='iterationId'),
tag_ids: Optional[TagIds5] = Query(None, alias='tagIds'),
):
"""
Gets the number of images tagged with the provided {tagIds} that have prediction results from
training for the provided iteration {iterationId}.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.delete(
'/projects/{projectId}/iterations/{iterationId}/publish',
tags=['iteration_control', 'project_lifecycle_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def unpublish_iteration(
project_id: UUID = Path(..., alias='projectId'),
iteration_id: UUID = Path(..., alias='iterationId'),
):
"""
Unpublish a specific iteration.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/projects/{projectId}/iterations/{iterationId}/publish',
tags=['iteration_control', 'project_lifecycle_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def publish_iteration(
project_id: UUID = Path(..., alias='projectId'),
iteration_id: UUID = Path(..., alias='iterationId'),
publish_name: str = Query(..., alias='publishName'),
prediction_id: str = Query(..., alias='predictionId'),
):
"""
Publish a specific iteration.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.delete(
'/projects/{projectId}/predictions',
tags=[
'project_lifecycle_management',
'image_handling_operations',
'prediction_management',
],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def delete_prediction(project_id: UUID = Path(..., alias='projectId'), ids: Ids = ...):
"""
Delete a set of predicted images and their associated prediction results.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/projects/{projectId}/predictions/query',
tags=['image_handling_operations', 'prediction_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def query_predictions(
project_id: UUID = Path(..., alias='projectId'), body: PredictionQueryToken = ...
):
"""
Get images that were sent to your prediction endpoint.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/projects/{projectId}/quicktest/image',
tags=['image_handling_operations', 'image_testing_functions'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def quick_test_image(
project_id: UUID = Path(..., alias='projectId'),
iteration_id: Optional[UUID] = Query(None, alias='iterationId'),
store: Optional[bool] = True,
file: UploadFile = ...,
):
"""
Quick test an image.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/projects/{projectId}/quicktest/url',
tags=['image_testing_functions', 'image_handling_operations'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def quick_test_image_url(
project_id: UUID = Path(..., alias='projectId'),
iteration_id: Optional[UUID] = Query(None, alias='iterationId'),
store: Optional[bool] = True,
body: ImageUrl = ...,
):
"""
Quick test an image url.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/projects/{projectId}/tags',
tags=[
'project_lifecycle_management',
'tagging_system_management',
'iteration_control',
],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def get_tags(
project_id: UUID = Path(..., alias='projectId'),
iteration_id: Optional[UUID] = Query(None, alias='iterationId'),
):
"""
Get the tags for a given project and iteration.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/projects/{projectId}/tags',
tags=['project_lifecycle_management', 'tagging_system_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def create_tag(
project_id: UUID = Path(..., alias='projectId'),
name: str = ...,
description: Optional[str] = None,
type: Optional[Type1] = None,
):
"""
Create a tag for the project.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.delete(
'/projects/{projectId}/tags/{tagId}',
tags=['project_lifecycle_management', 'tagging_system_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def delete_tag(
project_id: UUID = Path(..., alias='projectId'),
tag_id: UUID = Path(..., alias='tagId'),
):
"""
Delete a tag from the project.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/projects/{projectId}/tags/{tagId}',
tags=['tagging_system_management', 'project_lifecycle_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def get_tag(
project_id: UUID = Path(..., alias='projectId'),
tag_id: UUID = Path(..., alias='tagId'),
iteration_id: Optional[UUID] = Query(None, alias='iterationId'),
):
"""
Get information about a specific tag.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.patch(
'/projects/{projectId}/tags/{tagId}',
tags=['tagging_system_management', 'project_lifecycle_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def update_tag(
project_id: UUID = Path(..., alias='projectId'),
tag_id: UUID = Path(..., alias='tagId'),
body: Tag = ...,
):
"""
Update a tag.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/projects/{projectId}/tagsandregions/suggestions',
description=""" This API will get suggested tags and regions for an array/batch of untagged images along with confidences for the tags. It returns an empty array if no tags are found.
There is a limit of 64 images in the batch. """,
tags=['image_handling_operations', 'tagging_system_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def suggest_tags_and_regions(
project_id: UUID = Path(..., alias='projectId'),
iteration_id: UUID = Query(..., alias='iterationId'),
image_ids: ImageIds2 = Query(..., alias='imageIds'),
):
"""
Suggest tags and regions for an array/batch of untagged images. Returns empty array if no tags are found.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/projects/{projectId}/train',
tags=['training_process_management', 'project_lifecycle_management'],
security=[
APIKeyHeader(name="Training-Key"),
],
)
def train_project(
project_id: UUID = Path(..., alias='projectId'),
training_type: Optional[TrainingType] = Query(None, alias='trainingType'),
reserved_budget_in_hours: Optional[int] = Query(0, alias='reservedBudgetInHours'),
force_train: Optional[bool] = Query(False, alias='forceTrain'),
notification_email_address: Optional[str] = Query(
None, alias='notificationEmailAddress'
),
body: TrainingParameters = None,
):
"""
Queues project for training.
"""
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)