# generated by fastapi-codegen: # filename: openapi.yaml # timestamp: 2025-06-29T03:47:48+00:00 import argparse import json import os from typing import * from autogen.mcp.mcp_proxy import MCPProxy from autogen.mcp.mcp_proxy.security import APIKeyHeader, BaseSecurity from fastapi import Path from models import ( AccountsAccountOrderImpactPostRequest, AccountsAccountOrderImpactPostResponse, AccountsAccountOrdersCustomerOrderIdDeleteResponse, AccountsAccountOrdersCustomerOrderIdGetResponse, AccountsAccountOrdersCustomerOrderIdPutRequest, AccountsAccountOrdersCustomerOrderIdPutResponse, AccountsAccountOrdersGetResponse, AccountsAccountOrdersPostRequest, AccountsAccountOrdersPostResponse, AccountsAccountPositionsGetResponse, AccountsAccountSummaryGetResponse, AccountsAccountTradesGetResponse, AccountsGetResponse, MarketdataExchangeComponentsGetResponse, MarketdataSnapshotGetRequest, MarketdataSnapshotGetResponse, OauthAccessTokenPostRequest, OauthAccessTokenPostResponse, OauthLiveSessionTokenPostRequest, OauthLiveSessionTokenPostResponse, OauthRequestTokenPostRequest, OauthRequestTokenPostResponse, SecdefGetRequest, SecdefGetResponse, ) app = MCPProxy( description='Interactive Brokers Web API for 3rd Party Companies', title='IBKR 3rd Party Web API', version='1.0.0', servers=[{'url': 'https://www.interactivebrokers.com/tradingapi/v1'}], ) @app.get( '/accounts', description=""" Allows the caller to request a list of accounts associated with the session. """, tags=['account_management', 'trade_activity_tracking'], security=[ APIKeyHeader(name="portal"), ], ) def get_accounts(account: str): """ Brokerage Accounts """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/accounts/{account}/order_impact', description=""" This endpoint allows the consumer to check the impact that an order would have on the account, including margin, NLV and estimated commission costs. To specify the contract, you provide a value for the ContractId field, OR Ticker/ListingExchange/InstrumentType=STK for stocks OR Ticker/Currency/InstrumentType=CASH for FX. """, tags=['account_management', 'trade_activity_tracking'], security=[ APIKeyHeader(name="portal"), ], ) def post_accounts__account_order_impact( account: str, body: AccountsAccountOrderImpactPostRequest = ... ): """ Return margin impact info """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/accounts/{account}/orders', description=""" Returns a list of orders for the account passed in the URL """, tags=['account_management', 'trade_activity_tracking'], security=[ APIKeyHeader(name="portal"), ], ) def get_accounts__account_orders(account: str): """ Open Orders """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/accounts/{account}/orders', description=""" Places order """, tags=['account_management', 'trade_activity_tracking'], security=[ APIKeyHeader(name="portal"), ], ) def post_accounts__account_orders( account: str, body: AccountsAccountOrdersPostRequest = ... ): """ Place Order """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/accounts/{account}/orders/{CustomerOrderId}', description=""" Cancels the order with the referenced Customer Order ID for the account passed in the URL. """, tags=['account_management'], security=[ APIKeyHeader(name="portal"), ], ) def delete_accounts__account_orders___customer_order_id( account: str, customer_order_id: str = Path(..., alias='CustomerOrderId') ): """ Cancel Order """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/accounts/{account}/orders/{CustomerOrderId}', description=""" Returns the order with the referenced Customer Order ID for the account passed in the URL. """, tags=['account_management', 'trade_activity_tracking'], security=[ APIKeyHeader(name="portal"), ], ) def get_accounts__account_orders___customer_order_id( account: str, customer_order_id: str = Path(..., alias='CustomerOrderId') ): """ Return specific order info """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.put( '/accounts/{account}/orders/{CustomerOrderId}', description=""" Allows the caller to modify the order with the referenced Customer Order ID specified in the URL. A separate Customer Order ID must be provided in the request body for the modification. """, tags=['account_management', 'trade_activity_tracking'], security=[ APIKeyHeader(name="portal"), ], ) def put_accounts__account_orders___customer_order_id( account: str, customer_order_id: str = Path(..., alias='CustomerOrderId'), body: AccountsAccountOrdersCustomerOrderIdPutRequest = ..., ): """ Modify Order """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/accounts/{account}/positions', description=""" Returns a list of positions for the indicated account. """, tags=['account_management', 'trade_activity_tracking'], security=[ APIKeyHeader(name="portal"), ], ) def get_accounts__account_positions(account: str): """ Account Positions """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/accounts/{account}/summary', description=""" Returns a list of account and margin balances associated with the account passed in the URL """, tags=['account_management', 'trade_activity_tracking'], security=[ APIKeyHeader(name="portal"), ], ) def get_accounts__account_summary(account: str): """ Account Values Summary """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/accounts/{account}/trades', description=""" Returns a list of trades for the account starting at the given 'since' date to the current time (now()). Timezone is UTC. Any request with a future since date or going further than one week will result in an HTTP 400 bad request response. Calling /trades without since will return all trades for the past 24 hours. """, tags=['account_management', 'trade_activity_tracking'], security=[ APIKeyHeader(name="portal"), ], ) def get_accounts__account_trades(account: str, body: str = None): """ Returns trades in account """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/marketdata/exchange_components', description=""" This endpoint provides a bit mapping for the bid/ask/last 'market' values in the snapshot response. """, tags=['market_data_extraction'], security=[ APIKeyHeader(name="portal"), ], ) def get_marketdata_exchange_components(): """ Exchange Components """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/marketdata/snapshot', description=""" This endpoint allows the consumer to request a market data snapshot for one or more trading products. Consumers need to provide unique identifiers (conids) for the products in the IB product database (retrievable using the /secdef endpoint). The 'market' values are integers whose bits indicate the exchange(s) making up the quote. The mapping of bit to exchange is obtained from the marketdata/exchange_component endpoint. For example, if a bid has a 'market' value of 5 and the exchange_component result has the map 0 => NYSE, 1 => ISLAND, 2 => ARCA then the exchanges contributing to the bid size are NYSE and ARCA. Similarly, if market=2, then only ISLAND is contributing. """, tags=['market_data_extraction'], security=[ APIKeyHeader(name="portal"), ], ) def get_marketdata_snapshot(body: MarketdataSnapshotGetRequest): """ Market Data Snapshot """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/oauth/access_token', description=""" Obtain an access token using the request token and the verification code you received after the user provided authorization. See section 6.3 of the OAuth v1.0a specification for more details. """, tags=['oauth_token_management'], security=[ APIKeyHeader(name="portal"), ], ) def post_oauth_access_token(body: OauthAccessTokenPostRequest): """ Obtain a access token """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/oauth/live_session_token', description=""" In order to access protected IB resources, a live session token is required. This endpoint allows consumers to obtain a live session token to access these resources using an OAuth access token and the Diffie-Hellman prime and generator supplied during the registration process. Note this is an additional IB-specific step, and not part of the OAuth v1.0a specification. Please refer to the "OAuth at Interactive Brokers" document for more details. https://www.interactivebrokers.com/webtradingapi/oauth.pdf """, tags=['oauth_token_management'], security=[ APIKeyHeader(name="portal"), ], ) def post_oauth_live_session_token(body: OauthLiveSessionTokenPostRequest): """ Obtain a live session token """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/oauth/request_token', description=""" Obtain a request token. See section 6.1 of the OAuth v1.0a specification for more information. http://oauth.net/core/1.0a/#auth_step1 Note we do not return an oauth_token_secret in the response as we are using RSA signatures rather than PLAINTEXT authentication. """, tags=['oauth_token_management'], security=[ APIKeyHeader(name="portal"), ], ) def post_oauth_request_token(body: OauthRequestTokenPostRequest): """ Obtain a request token """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/secdef', description=""" Fields that compose security definition. Allowed combinations, (1) type and symbol and currency, or (2) type, symbol, exchange, and currency, or (3) conid """, tags=['market_data_extraction'], security=[ APIKeyHeader(name="portal"), ], ) def get_secdef(body: SecdefGetRequest): """ Get security definition """ 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)