# generated by fastapi-codegen:
# filename: openapi.yaml
# timestamp: 2025-06-29T12:06:32+00:00
import argparse
import json
import os
from typing import *
from typing import Optional, Union
from autogen.mcp.mcp_proxy import MCPProxy
from autogen.mcp.mcp_proxy.security import APIKeyHeader, BaseSecurity
from fastapi import Header, Path, Query
from models import (
ApiRnsPubCyclesGetResponse,
ApiRnsPubSubscriptionsGetResponse,
ApiRnsPubSubscriptionsSubscriptionIdConversationMessageGetResponse,
ApiRnsPvtPlansGetResponse,
ApiRnsPvtReportsGetResponse,
ReportResponse,
Settings,
SimulateResponseVO,
StorePlan,
SubscriptionCycleResponse,
SubscriptionGroupRequest,
SubscriptionGroupResponse,
SubscriptionThinItemRequest,
SubscriptionUpdateRequestV3,
UpdateItemInput,
)
app = MCPProxy(
contact={},
description="\r\nA **Subscription** is a list of items (SKUs) tied to certain recurring purchase settings:\r\n\r\n- User profile\r\n- Address\r\n- Payment method\r\n- Frequency\r\n- Cycle\r\n\r\nOnce you have [configured subscriptions](https://help.vtex.com/tutorial/how-to-configure-subscriptions%20--1FA9dfE7vJqxBna9Nft5Sj) in your store, the Subscriptions API allows you to create, manage and monitor your customers' subscriptions.\r\n\r\n\r\n\r\nTo read more about the Subscriptions feature, check our article [How Subscription works](https://help.vtex.com/tutorial/how-subscriptions-work--frequentlyAskedQuestions_4453).",
title='Subscriptions API (v3)',
version='1.0',
servers=[
{'url': 'https://vtex.local'},
{
'description': 'VTEX server url',
'url': 'https://{accountName}.{environment}.com.br',
'variables': {
'accountName': {
'default': '{accountName}',
'description': 'Name of the VTEX account. Used as part of the URL.',
},
'environment': {
'default': '{environment}',
'description': 'Environment to use. Used as part of the URL.',
},
},
},
],
)
@app.get(
'/api/rns/pub/cycles',
description=""" List cycles filtering by some arguments. """,
tags=['cycle_management', 'subscription_management'],
security=[
APIKeyHeader(name="X-VTEX-API-AppKey"),
APIKeyHeader(name="X-VTEX-API-AppToken"),
],
)
def get_api_rns_pub_cycles(
begin_date: Optional[str] = Query(None, alias='beginDate'),
end_date: Optional[str] = Query(None, alias='endDate'),
subscription_id: Optional[str] = Query(None, alias='subscriptionId'),
customer_email: Optional[str] = Query(None, alias='customerEmail'),
status: Optional[str] = None,
page: Optional[int] = 1,
size: Optional[int] = 15,
content__type: str = Header(..., alias='Content-Type'),
accept: str = Header(..., alias='Accept'),
):
"""
List cycles
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/api/rns/pub/cycles/{cycleId}',
description=""" Retrieve a specific cycle by its ID. """,
tags=['cycle_management'],
security=[
APIKeyHeader(name="X-VTEX-API-AppKey"),
APIKeyHeader(name="X-VTEX-API-AppToken"),
],
)
def get_api_rns_pub_cycles__cycle_id(
cycle_id: str = Path(..., alias='cycleId'),
content__type: str = Header(..., alias='Content-Type'),
accept: str = Header(..., alias='Accept'),
):
"""
Get cycle details
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/api/rns/pub/cycles/{cycleId}/retry',
description=""" Every subscription order has an execution count called cycle, which determines the position of an order counting from when the shopper subscribed. This endpoint reruns a cycle that is currently in error state. """,
tags=['cycle_management'],
security=[
APIKeyHeader(name="X-VTEX-API-AppKey"),
APIKeyHeader(name="X-VTEX-API-AppToken"),
],
)
def post_api_rns_pub_cycles__cycle_id_retry(
cycle_id: str = Path(..., alias='cycleId'),
content__type: str = Header(..., alias='Content-Type'),
accept: str = Header(..., alias='Accept'),
):
"""
Retry cycle
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/api/rns/pub/subscriptions',
description=""" List subscriptions filtering by some arguments. """,
tags=['subscription_management', 'plan_management'],
security=[
APIKeyHeader(name="X-VTEX-API-AppKey"),
APIKeyHeader(name="X-VTEX-API-AppToken"),
],
)
def get_api_rns_pub_subscriptions(
customer_email: Optional[str] = Query(None, alias='customerEmail'),
status: Optional[str] = None,
address_id: Optional[str] = Query(None, alias='addressId'),
payment_id: Optional[str] = Query(None, alias='paymentId'),
plan_id: Optional[str] = Query(None, alias='planId'),
next_purchase_date: Optional[str] = Query(None, alias='nextPurchaseDate'),
original_order_id: Optional[str] = Query(None, alias='originalOrderId'),
page: Optional[int] = 1,
size: Optional[int] = 15,
content__type: str = Header(..., alias='Content-Type'),
accept: str = Header(..., alias='Accept'),
):
"""
List subscriptions
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/api/rns/pub/subscriptions',
description=""" Create a new subscription. """,
tags=['subscription_management'],
security=[
APIKeyHeader(name="X-VTEX-API-AppKey"),
APIKeyHeader(name="X-VTEX-API-AppToken"),
],
)
def post_api_rns_pub_subscriptions(
content__type: str = Header(..., alias='Content-Type'),
accept: str = Header(..., alias='Accept'),
body: SubscriptionGroupRequest = None,
):
"""
Create subscription
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/api/rns/pub/subscriptions/simulate',
description=""" Simulates an order made by subscriptions on checkout and retrieves the current price for items and shipping. """,
tags=['subscription_management', 'plan_management'],
security=[
APIKeyHeader(name="X-VTEX-API-AppKey"),
APIKeyHeader(name="X-VTEX-API-AppToken"),
],
)
def post_api_rns_pub_subscriptions_simulate(
content__type: str = Header(..., alias='Content-Type'),
accept: str = Header(..., alias='Accept'),
body: SubscriptionGroupRequest = None,
):
"""
Calculate the current prices for the provided subscription template
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/api/rns/pub/subscriptions/{id}',
description=""" Retrieve a specific subscription by its ID. """,
tags=['subscription_management'],
security=[
APIKeyHeader(name="X-VTEX-API-AppKey"),
APIKeyHeader(name="X-VTEX-API-AppToken"),
],
)
def get_api_rns_pub_subscriptions__id(
id: str,
content__type: str = Header(..., alias='Content-Type'),
accept: str = Header(..., alias='Accept'),
):
"""
Get subscription details
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.patch(
'/api/rns/pub/subscriptions/{id}',
description=""" Update a specific subscription. """,
tags=['subscription_management'],
security=[
APIKeyHeader(name="X-VTEX-API-AppKey"),
APIKeyHeader(name="X-VTEX-API-AppToken"),
],
)
def patch_api_rns_pub_subscriptions__id(
id: str,
content__type: str = Header(..., alias='Content-Type'),
accept: str = Header(..., alias='Accept'),
body: SubscriptionUpdateRequestV3 = None,
):
"""
Update subscription
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/api/rns/pub/subscriptions/{id}/items',
description=""" Add a new item to a given subscription. """,
tags=['subscription_management'],
security=[
APIKeyHeader(name="X-VTEX-API-AppKey"),
APIKeyHeader(name="X-VTEX-API-AppToken"),
],
)
def post_api_rns_pub_subscriptions__id_items(
id: str,
content__type: str = Header(..., alias='Content-Type'),
accept: str = Header(..., alias='Accept'),
body: SubscriptionThinItemRequest = None,
):
"""
Add item to subscription
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.delete(
'/api/rns/pub/subscriptions/{id}/items/{itemId}',
description=""" Removes a specific item from a given subscription """,
tags=['subscription_management'],
security=[
APIKeyHeader(name="X-VTEX-API-AppKey"),
APIKeyHeader(name="X-VTEX-API-AppToken"),
],
)
def delete_api_rns_pub_subscriptions__id_items__item_id(
id: str,
item_id: str = Path(..., alias='itemId'),
content__type: str = Header(..., alias='Content-Type'),
accept: str = Header(..., alias='Accept'),
):
"""
Remove items from a subscription.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.patch(
'/api/rns/pub/subscriptions/{id}/items/{itemId}',
description=""" Edit a given item on a specific subscription """,
tags=['subscription_management'],
security=[
APIKeyHeader(name="X-VTEX-API-AppKey"),
APIKeyHeader(name="X-VTEX-API-AppToken"),
],
)
def patch_api_rns_pub_subscriptions__id_items__item_id(
id: str,
item_id: str = Path(..., alias='itemId'),
content__type: str = Header(..., alias='Content-Type'),
accept: str = Header(..., alias='Accept'),
body: UpdateItemInput = None,
):
"""
Edit items on a subscription.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/api/rns/pub/subscriptions/{id}/simulate',
description=""" Simulates an order made by the specific subscription on checkout and retrieves the current price for items and shipping. """,
tags=['subscription_management'],
security=[
APIKeyHeader(name="X-VTEX-API-AppKey"),
APIKeyHeader(name="X-VTEX-API-AppToken"),
],
)
def post_api_rns_pub_subscriptions__id_simulate(
id: str,
content__type: str = Header(..., alias='Content-Type'),
accept: str = Header(..., alias='Accept'),
):
"""
Calculate the current prices for a specific subscription
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/api/rns/pub/subscriptions/{subscriptionId}/conversation-message',
description=""" Retrieve all conversation messages sent to a customer regarding a given subscription. """,
tags=['conversation_management'],
security=[
APIKeyHeader(name="X-VTEX-API-AppKey"),
APIKeyHeader(name="X-VTEX-API-AppToken"),
],
)
def get_conversation_messages(
subscription_id: str = Path(..., alias='subscriptionId'),
content__type: str = Header(..., alias='Content-Type'),
accept: str = Header(..., alias='Accept'),
):
"""
Get conversation messages
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/api/rns/pvt/plans',
description=""" List plans filtering by some arguments. """,
tags=['plan_management', 'subscription_management'],
security=[
APIKeyHeader(name="X-VTEX-API-AppKey"),
APIKeyHeader(name="X-VTEX-API-AppToken"),
],
)
def get_api_rns_pvt_plans(
periodicity: Optional[str] = None,
interval: Optional[str] = None,
page: Optional[int] = 1,
size: Optional[int] = 15,
content__type: str = Header(..., alias='Content-Type'),
accept: str = Header(..., alias='Accept'),
):
"""
List plans
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/api/rns/pvt/plans/{id}',
description=""" This endpoint retrieves a specific plan by its ID. """,
tags=['plan_management'],
security=[
APIKeyHeader(name="X-VTEX-API-AppKey"),
APIKeyHeader(name="X-VTEX-API-AppToken"),
],
)
def get_api_rns_pvt_plans__id(
id: str,
content__type: str = Header(..., alias='Content-Type'),
accept: str = Header(..., alias='Accept'),
):
"""
Get plan details
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/api/rns/pvt/reports',
description=""" List all report templates available. """,
tags=['report_generation'],
security=[
APIKeyHeader(name="X-VTEX-API-AppKey"),
APIKeyHeader(name="X-VTEX-API-AppToken"),
],
)
def get_api_rns_pvt_reports(
content__type: str = Header(..., alias='Content-Type'),
accept: str = Header(..., alias='Accept'),
):
"""
List report templates
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/api/rns/pvt/reports/{reportName}/documents',
description=""" This endpoint creates a new report in the format of a CSV file and sends it via email. You can generate one of the following reports:
- subscriptionsWithStatus
- subscriptionsScheduledBetweenDate
- subscriptionsUpdatedBetweenDate
- subscriptionsCreatedBetweenDate
- executionsBetweenDate """,
tags=['report_generation'],
security=[
APIKeyHeader(name="X-VTEX-API-AppKey"),
APIKeyHeader(name="X-VTEX-API-AppToken"),
],
)
def post_api_rns_pvt_reports__report_name_documents(
report_name: str = Path(..., alias='reportName'),
email: Optional[str] = 'receiver@email.com',
begin_date: Optional[str] = Query('2022-09-01', alias='beginDate'),
end_date: Optional[str] = Query('2022-10-01', alias='endDate'),
content__type: str = Header(..., alias='Content-Type'),
accept: str = Header(..., alias='Accept'),
):
"""
Generate report
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/api/rns/pvt/reports/{reportName}/documents/{documentId}',
description=""" Retrieve a specific report document by its Id. """,
tags=['report_generation'],
security=[
APIKeyHeader(name="X-VTEX-API-AppKey"),
APIKeyHeader(name="X-VTEX-API-AppToken"),
],
)
def get_api_rns_pvt_reports__report_name_documents__document_id(
report_name: str = Path(..., alias='reportName'),
document_id: str = Path(..., alias='documentId'),
content__type: str = Header(..., alias='Content-Type'),
accept: str = Header(..., alias='Accept'),
):
"""
Get report document details
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/api/rns/settings',
description=""" List the details of the settings of a given subscription. """,
tags=['subscription_management', 'settings_management'],
security=[
APIKeyHeader(name="X-VTEX-API-AppKey"),
APIKeyHeader(name="X-VTEX-API-AppToken"),
],
)
def get_settings(
content__type: str = Header(..., alias='Content-Type'),
accept: str = Header(..., alias='Accept'),
):
"""
Get Subscription Settings
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/api/rns/settings',
description=""" Edits Subscriptions settings in your store. """,
tags=['subscription_management', 'settings_management'],
security=[
APIKeyHeader(name="X-VTEX-API-AppKey"),
APIKeyHeader(name="X-VTEX-API-AppToken"),
],
)
def edit_settings(
content__type: str = Header(..., alias='Content-Type'),
accept: str = Header(..., alias='Accept'),
body: Settings = ...,
):
"""
Edit Subscriptions settings
"""
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)