main.py•11.6 kB
# generated by fastapi-codegen:
# filename: openapi.yaml
# timestamp: 2025-07-16T00:44:01+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,
APIKeyQuery,
BaseSecurity,
UnsuportedSecurityStub,
)
from fastapi import Path, Query
from models import (
BatchInputMarketingEventCreateRequestParams,
BatchInputMarketingEventEmailSubscriber,
BatchInputMarketingEventExternalUniqueIdentifier,
BatchInputMarketingEventSubscriber,
BatchResponseMarketingEventPublicDefaultResponse,
BatchResponseSubscriberEmailResponse,
BatchResponseSubscriberVidResponse,
CollectionResponseMarketingEventExternalUniqueIdentifierNoPaging,
Error,
EventDetailSettings,
EventDetailSettingsUrl,
MarketingEventCompleteRequestParams,
MarketingEventCreateRequestParams,
MarketingEventDefaultResponse,
MarketingEventPublicDefaultResponse,
MarketingEventPublicReadResponse,
MarketingEventUpdateRequestParams,
)
app = MCPProxy(
description="These APIs allow you to interact with HubSpot's Marketing Events Extension. It allows you to:\n* Create, Read or update Marketing Event information in HubSpot\n* Specify whether a HubSpot contact has registered, attended or cancelled a registration to a Marketing Event.\n* Specify a URL that can be called to get the details of a Marketing Event.\n",
title='Marketing Events Extension',
version='v3',
servers=[{'url': 'https://api.hubapi.com/'}],
)
@app.post(
'/marketing/v3/marketing-events/attendance/{externalEventId}/{subscriberState}/create',
description=""" Record a subscription state between multiple HubSpot contacts and a marketing event, using HubSpot contact ids. """,
tags=['marketing_event_attendance_logging', 'marketing_event_subscriber_handling'],
security=[
APIKeyHeader(name="private-app-legacy"),
UnsuportedSecurityStub(name="None"),
],
)
def record_marketing_event_attendance(
external_event_id: str = Path(..., alias='externalEventId'),
subscriber_state: str = Path(..., alias='subscriberState'),
external_account_id: Optional[str] = Query(None, alias='externalAccountId'),
body: BatchInputMarketingEventSubscriber = ...,
):
"""
Record
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/marketing/v3/marketing-events/attendance/{externalEventId}/{subscriberState}/email-create',
description=""" Record a subscription state between multiple HubSpot contacts and a marketing event, using contact email addresses. If contact is not present it will be automatically created. If you set params """,
tags=['marketing_event_attendance_logging', 'marketing_event_subscriber_handling'],
security=[
APIKeyHeader(name="private-app-legacy"),
UnsuportedSecurityStub(name="None"),
],
)
def log_marketing_event_attendance(
external_event_id: str = Path(..., alias='externalEventId'),
subscriber_state: str = Path(..., alias='subscriberState'),
external_account_id: Optional[str] = Query(None, alias='externalAccountId'),
body: BatchInputMarketingEventEmailSubscriber = ...,
):
"""
Record
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/marketing/v3/marketing-events/events',
tags=['marketing_event_crud_operations'],
security=[
APIKeyHeader(name="private-app-legacy"),
UnsuportedSecurityStub(name="None"),
],
)
def post__marketing_v3_marketing_events_events_create(
body: MarketingEventCreateRequestParams,
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/marketing/v3/marketing-events/events/delete',
tags=['marketing_event_attendance_logging', 'marketing_event_subscriber_handling'],
security=[
APIKeyHeader(name="private-app-legacy"),
UnsuportedSecurityStub(name="None"),
],
)
def post__marketing_v3_marketing_events_events_delete_archive_batch(
body: BatchInputMarketingEventExternalUniqueIdentifier,
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/marketing/v3/marketing-events/events/search',
description=""" Search for marketing events that have an event id that starts with the query string """,
tags=['marketing_event_query'],
security=[
UnsuportedSecurityStub(name="None"),
APIKeyHeader(name="private-app-legacy"),
],
)
def get__marketing_v3_marketing_events_events_search_do_search(q: str):
"""
Search for marketing events
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/marketing/v3/marketing-events/events/upsert',
tags=['marketing_event_crud_operations', 'marketing_event_subscriber_handling'],
security=[
APIKeyHeader(name="private-app-legacy"),
UnsuportedSecurityStub(name="None"),
],
)
def post__marketing_v3_marketing_events_events_upsert_do_upsert(
body: BatchInputMarketingEventCreateRequestParams,
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.delete(
'/marketing/v3/marketing-events/events/{externalEventId}',
tags=['marketing_event_attendance_logging'],
security=[
APIKeyHeader(name="private-app-legacy"),
UnsuportedSecurityStub(name="None"),
],
)
def archive_marketing_event_by_ids(
external_event_id: str = Path(..., alias='externalEventId'),
external_account_id: str = Query(..., alias='externalAccountId'),
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/marketing/v3/marketing-events/events/{externalEventId}',
tags=['marketing_event_attendance_logging', 'marketing_event_subscriber_handling'],
security=[
UnsuportedSecurityStub(name="None"),
APIKeyHeader(name="private-app-legacy"),
],
)
def get_marketing_event_by_external_id_and_account_id(
external_event_id: str = Path(..., alias='externalEventId'),
external_account_id: str = Query(..., alias='externalAccountId'),
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.patch(
'/marketing/v3/marketing-events/events/{externalEventId}',
tags=['marketing_event_crud_operations', 'marketing_event_subscriber_handling'],
security=[
APIKeyHeader(name="private-app-legacy"),
UnsuportedSecurityStub(name="None"),
],
)
def update_marketing_event_by_external_id(
external_event_id: str = Path(..., alias='externalEventId'),
external_account_id: str = Query(..., alias='externalAccountId'),
body: MarketingEventUpdateRequestParams = ...,
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.put(
'/marketing/v3/marketing-events/events/{externalEventId}',
tags=['marketing_event_crud_operations', 'marketing_event_subscriber_handling'],
security=[
APIKeyHeader(name="private-app-legacy"),
UnsuportedSecurityStub(name="None"),
],
)
def replace_external_event_id_with_marketing_event(
external_event_id: str = Path(..., alias='externalEventId'),
body: MarketingEventCreateRequestParams = ...,
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/marketing/v3/marketing-events/events/{externalEventId}/cancel',
tags=['marketing_event_attendance_logging', 'marketing_event_subscriber_handling'],
security=[
APIKeyHeader(name="private-app-legacy"),
UnsuportedSecurityStub(name="None"),
],
)
def cancel_marketing_event_by_external_id(
external_event_id: str = Path(..., alias='externalEventId'),
external_account_id: str = Query(..., alias='externalAccountId'),
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/marketing/v3/marketing-events/events/{externalEventId}/complete',
tags=['marketing_event_crud_operations'],
security=[
APIKeyHeader(name="private-app-legacy"),
UnsuportedSecurityStub(name="None"),
],
)
def complete_marketing_event_by_external_id(
external_event_id: str = Path(..., alias='externalEventId'),
external_account_id: str = Query(..., alias='externalAccountId'),
body: MarketingEventCompleteRequestParams = ...,
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/marketing/v3/marketing-events/events/{externalEventId}/{subscriberState}/email-upsert',
tags=['marketing_event_attendance_logging', 'marketing_event_subscriber_handling'],
security=[
APIKeyHeader(name="private-app-legacy"),
UnsuportedSecurityStub(name="None"),
],
)
def upsert_email_subscriber_by_external_event_id(
external_event_id: str = Path(..., alias='externalEventId'),
subscriber_state: str = Path(..., alias='subscriberState'),
external_account_id: str = Query(..., alias='externalAccountId'),
body: BatchInputMarketingEventEmailSubscriber = ...,
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/marketing/v3/marketing-events/events/{externalEventId}/{subscriberState}/upsert',
tags=['marketing_event_attendance_logging', 'marketing_event_subscriber_handling'],
security=[
APIKeyHeader(name="private-app-legacy"),
UnsuportedSecurityStub(name="None"),
],
)
def upsert_marketing_event_subscriber_by_external_id(
external_event_id: str = Path(..., alias='externalEventId'),
subscriber_state: str = Path(..., alias='subscriberState'),
external_account_id: str = Query(..., alias='externalAccountId'),
body: BatchInputMarketingEventSubscriber = ...,
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/marketing/v3/marketing-events/{appId}/settings',
tags=['application_settings_configuration'],
security=[
APIKeyQuery(name="hapikey"),
APIKeyQuery(name="hapikey"),
],
)
def get__marketing_v3_marketing_events__app_id__settings_get_all(
app_id: int = Path(..., alias='appId')
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/marketing/v3/marketing-events/{appId}/settings',
tags=['application_settings_configuration'],
security=[
APIKeyQuery(name="hapikey"),
APIKeyQuery(name="hapikey"),
],
)
def post__marketing_v3_marketing_events__app_id__settings_create(
app_id: int = Path(..., alias='appId'), body: EventDetailSettingsUrl = ...
):
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)