main.py•2.62 kB
# generated by fastapi-codegen:
# filename: openapi.yaml
# timestamp: 2025-06-29T07:57:41+00:00
import argparse
import json
import os
from typing import *
from typing import Union
from autogen.mcp.mcp_proxy import MCPProxy
from autogen.mcp.mcp_proxy.security import APIKeyHeader, BaseSecurity
from models import (
BulkMappingJob,
BulkMappingJobResult,
Key,
MappingValuesKeyGetResponse,
)
app = MCPProxy(
contact={
'email': 'support@openfigi.com',
'name': 'OpenFIGI API Support',
'url': 'https://openfigi.com/api',
},
description='A free & open API for FIGI discovery.',
license={
'name': 'Apache 2.0',
'url': 'https://www.apache.org/licenses/LICENSE-2.0.html',
},
termsOfService='https://openfigi.com/docs/terms-of-service',
title='OpenFIGI API',
version='1.4.0',
servers=[
{
'description': 'Production Tier',
'url': 'https://api.openfigi.com/{basePath}',
'variables': {'basePath': {'default': 'v1', 'enum': ['v1', 'v2', 'v3']}},
}
],
)
@app.post(
'/mapping',
description=""" Allows mapping from third-party identifiers to FIGIs. """,
tags=['mapping_job_management'],
security=[
APIKeyHeader(name="X-OPENFIGI-APIKEY"),
],
)
def post_mapping(body: BulkMappingJob = None):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/mapping/values/{key}',
description=""" Get values for enum-like fields. """,
tags=['mapping_job_management'],
security=[
APIKeyHeader(name="X-OPENFIGI-APIKEY"),
],
)
def get_mapping_values__key(key: Key):
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)