Schwab Model Context Protocol Server

import asyncio import datetime from enum import Enum from types import SimpleNamespace from typing import Any, cast from schwab.client import AsyncClient from schwab_mcp.context import SchwabContext, SchwabServerContext from schwab_mcp.tools import orders from schwab_mcp.approvals import ApprovalDecision, ApprovalManager, ApprovalRequest class DummyApprovalManager(ApprovalManager): async def require(self, request: ApprovalRequest) -> ApprovalDecision: # noqa: ARG002 return ApprovalDecision.APPROVED class DummyOrdersClient: class Order: Status = Enum( "Status", "WORKING FILLED CANCELED REJECTED", ) async def get_orders_for_account(self, *args, **kwargs): return None def run(coro): return asyncio.run(coro) def make_ctx(client: Any) -> SchwabContext: lifespan_context = SchwabServerContext( client=cast(AsyncClient, client), approval_manager=DummyApprovalManager(), ) request_context = SimpleNamespace(lifespan_context=lifespan_context) return SchwabContext.model_construct( _request_context=cast(Any, request_context), _fastmcp=None, ) def test_get_orders_maps_single_status_and_dates(monkeypatch): captured: dict[str, Any] = {} async def fake_call(func, *args, **kwargs): captured["func"] = func captured["args"] = args captured["kwargs"] = kwargs return "ok" monkeypatch.setattr(orders, "call", fake_call) client = DummyOrdersClient() ctx = make_ctx(client) result = run( orders.get_orders( ctx, "abc123", max_results=25, from_date="2024-04-01", to_date="2024-04-15", status="working", ) ) assert result == "ok" assert captured["func"] == client.get_orders_for_account args = captured["args"] assert isinstance(args, tuple) assert args == ("abc123",) kwargs = captured["kwargs"] assert isinstance(kwargs, dict) assert kwargs["max_results"] == 25 assert kwargs["from_entered_datetime"] == datetime.date(2024, 4, 1) assert kwargs["to_entered_datetime"] == datetime.date(2024, 4, 15) assert kwargs["status"] is client.Order.Status.WORKING def test_get_orders_maps_status_list(monkeypatch): captured: dict[str, Any] = {} async def fake_call(func, *args, **kwargs): captured["func"] = func captured["args"] = args captured["kwargs"] = kwargs return "ok" monkeypatch.setattr(orders, "call", fake_call) client = DummyOrdersClient() ctx = make_ctx(client) result = run( orders.get_orders( ctx, "xyz789", status=["filled", "canceled"], ) ) assert result == "ok" assert captured["func"] == client.get_orders_for_account args = captured["args"] assert isinstance(args, tuple) assert args == ("xyz789",) kwargs = captured["kwargs"] assert isinstance(kwargs, dict) assert "status" not in kwargs assert kwargs["statuses"] == [ client.Order.Status.FILLED, client.Order.Status.CANCELED, ] def test_place_equity_order_returns_order_metadata(): account_hash = "abc123" order_id = 987654321 location = ( f"https://api.schwabapi.com/trader/v1/accounts/{account_hash}/orders/{order_id}" ) class DummyResponse: status_code = 201 url = f"https://api.schwabapi.com/trader/v1/accounts/{account_hash}/orders" text = "" content = b"" headers = {"Location": location} is_error = False def raise_for_status(self) -> None: return None class DummyPlaceOrderClient(DummyOrdersClient): def __init__(self) -> None: self.captured: dict[str, Any] | None = None async def place_order(self, *args, **kwargs): self.captured = {"args": args, "kwargs": kwargs} return DummyResponse() client = DummyPlaceOrderClient() ctx = make_ctx(client) result = run( orders.place_equity_order( ctx, account_hash, "SPY", 1, "buy", "market", ) ) assert result == { "orderId": order_id, "accountHash": account_hash, "location": location, } captured = client.captured assert captured is not None assert captured["kwargs"]["account_hash"] == account_hash assert captured["kwargs"]["order_spec"]["orderStrategyType"] == "SINGLE"