# generated by fastapi-codegen:
# filename: openapi.yaml
# timestamp: 2025-06-29T09:31:31+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 APIKeyQuery, BaseSecurity, HTTPBearer
from pydantic import conint
from starlette.requests import Request
from models import (
Error,
Expand,
Fields,
Persona,
PersonasGetResponse,
Playlist,
PlaylistsGetResponse,
Show,
ShowsGetResponse,
Spin,
SpinsGetResponse,
SpinsPostResponse,
)
app = MCPProxy(
contact={'x-twitter': 'Spinitron'},
description="## Notes\n\n**Tutorial demo** using this API is at [https://spinitron.com/v2-api-demo/](https://spinitron.com/v2-api-demo/). For web integration using iframes and/or JavaScript instead of an API, see [https://spinitron.github.io/v2-web-integration/](https://spinitron.github.io/v2-web-integration/).\n\n**Your API key** is found in the Spinitron web app. Log in to Spinitron and go to *Automation & API* in the *Admin* menu.\n\n**Authenticate** by presenting your API key using either HTTP Bearer Authorization\n(preferred)\n\n curl -H 'Authorization: Bearer YOURAPIKEY' 'https://spinitron.com/api/spins'\n\nor in the query parameter `access-token` (less secure owing to webserver\nlog files)\n\n curl 'https://spinitron.com/api/spins?access-token=YOURAPIKEY'\n\n**Limit** per page of results is 20 by default and miximally 200.\n\n**Try it out** below works to\ngenerate example cURL requests but not to get responses from Spinitron. We\ndo not accept queries sent from web browsers. Copy-paste the cURL commands\nand run them from your computer.\n\n**Cache** the data you get from the API if you are using it in web or mobile integration. It's not ok to query the API on *every* page request you serve. The [demo](https://spinitron.com/v2-api-demo/) shows how easy it can be to implement a file cache.\n\nAn extension to this API with access to all stations for partner applications is available. Contact us.\n",
title='Spinitron v2 API',
version='1.0.0',
servers=[{'url': 'https://spinitron.com/api'}],
)
@app.get(
'/personas',
tags=['persona_management'],
security=[
HTTPBearer(name="None"),
APIKeyQuery(name="access-token"),
],
)
def get_personas(
name: Optional[str] = None,
count: Optional[conint(ge=1)] = 20,
page: Optional[conint(ge=1)] = None,
fields: Optional[Fields] = None,
expand: Optional[Expand] = None,
):
"""
Get Personas
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/personas/{id}',
tags=['persona_management'],
security=[
HTTPBearer(name="None"),
APIKeyQuery(name="access-token"),
],
)
def get_personas__id(
id: int, fields: Optional[Fields] = None, expand: Optional[Expand] = None
):
"""
Get Persona by id
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/playlists',
description=""" Get Playlists optionally filtered by a datetime range.
Only past Playlists will be returned (with allowed tolerance equals 1 hour in future).
Ordered chronologically from newest to oldest.
""",
tags=['playlist_handler', 'show_schedule_management'],
security=[
HTTPBearer(name="None"),
APIKeyQuery(name="access-token"),
],
)
def get_playlists(
start: Optional[datetime] = None,
end: Optional[datetime] = None,
show_id: Optional[int] = None,
persona_id: Optional[int] = None,
count: Optional[conint(ge=1)] = 20,
page: Optional[conint(ge=1)] = None,
fields: Optional[Fields] = None,
expand: Optional[Expand] = None,
):
"""
Returns playlists optionally filtered by {start} and/or {end} datetimes
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/playlists/{id}',
description=""" The response object represents the playlist specified by {id}.
Status 404 is returned if a playlist with {id} does not exist or if it does but starts in the future (with allowed tolerance equals 1 hour in future).
""",
tags=['playlist_handler'],
security=[
HTTPBearer(name="None"),
APIKeyQuery(name="access-token"),
],
)
def get_playlists__id(
id: int, fields: Optional[Fields] = None, expand: Optional[Expand] = None
):
"""
Get a Playlist by id
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/shows',
description=""" **Terminology**: Spinitron defines a *show* as a radio program. A show can have one or more *schedules*,
each of which may specify either an *occurence* or a *repetition*, which represents a set of occurences.
Thus scheduled shows have occurences that, for example, may be displayed in a calendar.
In the response, `items` is an array of objects representing occurences of scheduled shows.
You may optionally filter `items` to a datetime *range* by including in the request {start} and/or {end}
parameters, both of which must be no more than one hour in the past. An occurence starting at {end} is
included in the reponse.
`itmes` can include occurences that begin *or* end within the filter range. A show that goes on air before
{start} appears in `items` if it ends *after* but not *at* {start}. An occurence starting at or before {end}
is included.
If the request omits the {start} parameter, the server sets its value to the current time so that the filter
range's start is always defined. If the request specifies {end} then the requested range is *bounded*,
otherwise it is *unbounded*.
For a bounded request, `items` includes *every* occurence of all shows occuring in the range. The only
difference between objects in `items` representing a given show will be the `start` field value.
For an unbounded request, `items` includes *only one* occurence per show, specifically, the
next occurrence after {start} of all shows occuring after {start}.
Use an unbounded request to get a straight list all shows. Use a bounded request to get a calendar/agenda
of shows expanded into occurrences by thir shedules and repetitions.
Objects in `items` are ordered first by `datetime` and then by `id`.
""",
tags=['show_schedule_management'],
security=[
HTTPBearer(name="None"),
APIKeyQuery(name="access-token"),
],
)
def get_shows(
start: Optional[datetime] = None,
end: Optional[datetime] = None,
count: Optional[conint(ge=1)] = 20,
page: Optional[conint(ge=1)] = None,
fields: Optional[Fields] = None,
expand: Optional[Expand] = None,
):
"""
Returns scheduled shows optionally filtered by {start} and/or {end} datetimes
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/shows/{id}',
description=""" The response object represents the next occurence of the show specified by {id}.
Status 404 is returned if a show with {id} does not exist or if it does but all its scheduled occurences elapsed in the past.
""",
tags=['show_schedule_management'],
security=[
HTTPBearer(name="None"),
APIKeyQuery(name="access-token"),
],
)
def get_shows__id(
id: int, fields: Optional[Fields] = None, expand: Optional[Expand] = None
):
"""
Get a Show by id
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/spins',
description=""" Get Spins optionally filtered by a datetime range. Only past Spins will be returned.
""",
tags=['spin_event_logging', 'playlist_handler'],
security=[
HTTPBearer(name="None"),
APIKeyQuery(name="access-token"),
],
)
def get_spins(
start: Optional[datetime] = None,
end: Optional[datetime] = None,
playlist_id: Optional[int] = None,
show_id: Optional[int] = None,
count: Optional[conint(ge=1)] = 20,
page: Optional[conint(ge=1)] = None,
fields: Optional[Fields] = None,
expand: Optional[Expand] = None,
):
"""
Returns spins optionally filtered by {start} and/or {end} datetimes
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/spins',
description=""" An endpoint for automation systems to log spins into the spin table. """,
tags=['spin_event_logging'],
security=[
HTTPBearer(name="None"),
APIKeyQuery(name="access-token"),
],
)
def post_spins(request: Request):
"""
Log a Spin
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/spins/{id}',
tags=['spin_event_logging'],
security=[
HTTPBearer(name="None"),
APIKeyQuery(name="access-token"),
],
)
def get_spins__id(
id: int, fields: Optional[Fields] = None, expand: Optional[Expand] = None
):
"""
Get a Spin by id
"""
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)
