main.py•12.6 kB
# generated by fastapi-codegen:
# filename: openapi.yaml
# timestamp: 2025-07-07T16:18:59+00:00
import argparse
import json
import os
from typing import *
from typing import Optional, Union
from autogen.mcp.mcp_proxy import MCPProxy
from autogen.mcp.mcp_proxy.security import BaseSecurity, HTTPBearer
from fastapi import Header
from pydantic import constr
from models import (
CreateApiKeyRequest,
CreateApiKeyResponse,
CreateAudienceOptions,
CreateAudienceResponseSuccess,
CreateBatchEmailsResponse,
CreateBroadcastOptions,
CreateBroadcastResponseSuccess,
CreateContactOptions,
CreateContactResponseSuccess,
CreateDomainRequest,
CreateDomainResponse,
DeleteDomainResponse,
Domain,
Email,
EmailsBatchPostRequest,
GetAudienceResponseSuccess,
GetBroadcastResponseSuccess,
GetContactResponseSuccess,
ListApiKeysResponse,
ListAudiencesResponseSuccess,
ListBroadcastsResponseSuccess,
ListContactsResponseSuccess,
ListDomainsResponse,
RemoveAudienceResponseSuccess,
RemoveBroadcastResponseSuccess,
RemoveContactResponseSuccess,
SendBroadcastOptions,
SendBroadcastResponseSuccess,
SendEmailRequest,
SendEmailResponse,
UpdateContactOptions,
UpdateContactResponseSuccess,
UpdateDomainOptions,
UpdateDomainResponseSuccess,
UpdateEmailOptions,
VerifyDomainResponse,
)
app = MCPProxy(
description='Resend is the email platform for developers.',
title='Resend',
version='1.1.0',
servers=[{'url': 'https://api.resend.com'}],
)
@app.get(
'/api-keys',
tags=['api_key_management'],
security=[
HTTPBearer(name="None"),
],
)
def get_api_keys():
"""
Retrieve a list of API keys
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/api-keys',
tags=['api_key_management'],
security=[
HTTPBearer(name="None"),
],
)
def post_api_keys(body: CreateApiKeyRequest = None):
"""
Create a new API key
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.delete(
'/api-keys/{api_key_id}',
tags=['api_key_management'],
security=[
HTTPBearer(name="None"),
],
)
def delete_api_keys__api_key_id(api_key_id: str):
"""
Remove an existing API key
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/audiences',
tags=['audience_management'],
security=[
HTTPBearer(name="None"),
],
)
def get_audiences():
"""
Retrieve a list of audiences
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/audiences',
tags=['audience_management', 'contact_handling'],
security=[
HTTPBearer(name="None"),
],
)
def post_audiences(body: CreateAudienceOptions = None):
"""
Create a list of contacts
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/audiences/{audience_id}/contacts',
tags=['audience_management', 'contact_handling'],
security=[
HTTPBearer(name="None"),
],
)
def get_audiences__audience_id_contacts(audience_id: str):
"""
Retrieve a list of contacts
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/audiences/{audience_id}/contacts',
tags=['audience_management', 'contact_handling'],
security=[
HTTPBearer(name="None"),
],
)
def post_audiences__audience_id_contacts(
audience_id: str, body: CreateContactOptions = None
):
"""
Create a new contact
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.delete(
'/audiences/{audience_id}/contacts/{email}',
tags=['audience_management', 'contact_handling'],
security=[
HTTPBearer(name="None"),
],
)
def delete_audiences__audience_id_contacts__email(email: str, audience_id: str = ...):
"""
Remove an existing contact by email
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/audiences/{audience_id}/contacts/{email}',
tags=['audience_management', 'contact_handling'],
security=[
HTTPBearer(name="None"),
],
)
def get_audiences__audience_id_contacts__email(email: str, audience_id: str = ...):
"""
Retrieves a single contact by email
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.patch(
'/audiences/{audience_id}/contacts/{email}',
tags=['audience_management', 'contact_handling'],
security=[
HTTPBearer(name="None"),
],
)
def patch_audiences__audience_id_contacts__email(email: str, audience_id: str = ...):
"""
Update a single contact by email
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.delete(
'/audiences/{audience_id}/contacts/{id}',
tags=['audience_management', 'contact_handling'],
security=[
HTTPBearer(name="None"),
],
)
def delete_audiences__audience_id_contacts__id(id: str, audience_id: str = ...):
"""
Remove an existing contact by id
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/audiences/{audience_id}/contacts/{id}',
tags=['audience_management', 'contact_handling'],
security=[
HTTPBearer(name="None"),
],
)
def get_audiences__audience_id_contacts__id(id: str, audience_id: str = ...):
"""
Retrieve a single contact by id
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.patch(
'/audiences/{audience_id}/contacts/{id}',
tags=['audience_management', 'contact_handling'],
security=[
HTTPBearer(name="None"),
],
)
def patch_audiences__audience_id_contacts__id(
id: str, audience_id: str = ..., body: UpdateContactOptions = None
):
"""
Update a single contact by id
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.delete(
'/audiences/{id}',
tags=['audience_management'],
security=[
HTTPBearer(name="None"),
],
)
def delete_audiences__id(id: str):
"""
Remove an existing audience
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/audiences/{id}',
tags=['audience_management'],
security=[
HTTPBearer(name="None"),
],
)
def get_audiences__id(id: str):
"""
Retrieve a single audience
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/broadcasts',
tags=['broadcast_operations'],
security=[
HTTPBearer(name="None"),
],
)
def get_broadcasts():
"""
Retrieve a list of broadcasts
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/broadcasts',
tags=['broadcast_operations'],
security=[
HTTPBearer(name="None"),
],
)
def post_broadcasts(body: CreateBroadcastOptions = None):
"""
Create a broadcast
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.delete(
'/broadcasts/{id}',
tags=['broadcast_operations'],
security=[
HTTPBearer(name="None"),
],
)
def delete_broadcasts__id(id: str):
"""
Remove an existing broadcast that is in the draft status
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/broadcasts/{id}',
tags=['broadcast_operations'],
security=[
HTTPBearer(name="None"),
],
)
def get_broadcasts__id(id: str):
"""
Retrieve a single broadcast
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/broadcasts/{id}/send',
tags=['broadcast_operations', 'email_scheduling_management'],
security=[
HTTPBearer(name="None"),
],
)
def post_broadcasts__id_send(id: str, body: SendBroadcastOptions = None):
"""
Send or schedule a broadcast
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/domains',
tags=['domain_operations'],
security=[
HTTPBearer(name="None"),
],
)
def get_domains():
"""
Retrieve a list of domains
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/domains',
tags=['domain_operations'],
security=[
HTTPBearer(name="None"),
],
)
def post_domains(body: CreateDomainRequest = None):
"""
Create a new domain
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.delete(
'/domains/{domain_id}',
tags=['domain_operations'],
security=[
HTTPBearer(name="None"),
],
)
def delete_domains__domain_id(domain_id: str):
"""
Remove an existing domain
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/domains/{domain_id}',
tags=['domain_operations'],
security=[
HTTPBearer(name="None"),
],
)
def get_domains__domain_id(domain_id: str):
"""
Retrieve a single domain
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.patch(
'/domains/{domain_id}',
tags=['domain_operations'],
security=[
HTTPBearer(name="None"),
],
)
def patch_domains__domain_id(domain_id: str, body: UpdateDomainOptions = None):
"""
Update an existing domain
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/domains/{domain_id}/verify',
tags=['domain_operations'],
security=[
HTTPBearer(name="None"),
],
)
def post_domains__domain_id_verify(domain_id: str):
"""
Verify an existing domain
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/emails',
tags=['email_operations', 'broadcast_operations'],
security=[
HTTPBearer(name="None"),
],
)
def post_emails(
idempotency__key: Optional[constr(max_length=256)] = Header(
None, alias='Idempotency-Key'
),
body: SendEmailRequest = None,
):
"""
Send an email
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/emails/batch',
tags=['email_operations', 'broadcast_operations'],
security=[
HTTPBearer(name="None"),
],
)
def post_emails_batch(
idempotency__key: Optional[constr(max_length=256)] = Header(
None, alias='Idempotency-Key'
),
body: EmailsBatchPostRequest = None,
):
"""
Trigger up to 100 batch emails at once.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/emails/{email_id}',
tags=['email_operations'],
security=[
HTTPBearer(name="None"),
],
)
def get_emails__email_id(email_id: str):
"""
Retrieve a single email
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.patch(
'/emails/{email_id}',
tags=['contact_handling', 'email_operations'],
security=[
HTTPBearer(name="None"),
],
)
def patch_emails__email_id(email_id: str):
"""
Update a single email
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/emails/{email_id}/cancel',
tags=['email_scheduling_management', 'email_operations'],
security=[
HTTPBearer(name="None"),
],
)
def post_emails__email_id_cancel(email_id: str):
"""
Cancel the schedule of the e-mail.
"""
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)