Merge pull request modelcontextprotocol#26 from jkoelker/jk/technicals

feat(tools): add technical tools

Author: jkoelker

.github/workflows/ci.yml

@@ -1,16 +1,21 @@
+---
 name: CI
 
-on:
+"on":
   push:
-    branches: ["main"]
+    branches:
+      - main
   pull_request:
 
 jobs:
   lint-and-test:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.10", "3.11", "3.12"]
+        python-version:
+          - 3.10
+          - 3.11
+          - 3.12
     steps:
       - name: Checkout
         uses: actions/checkout@v4
@@ -21,7 +26,7 @@ jobs:
           python-version: ${{ matrix.python-version }}
 
       - name: Sync dependencies
-        run: uv sync --group dev
+        run: uv sync --group dev --group ta
 
       - name: Lint (ruff)
         run: uv run ruff check .
@@ -30,4 +35,4 @@ jobs:
         run: uv run pyright
 
       - name: Run tests
-        run: uv run python -m pytest
+        run: uv run pytest

pyproject.toml

@@ -32,6 +32,10 @@ dev = [
     "pytest>=8.4.2",
     "ruff>=0.14.0",
 ]
+ta = [
+    "pandas>=2.2.2",
+    "pandas-ta-classic>=0.3.36",
+]
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]

src/schwab_mcp/cli.py

@@ -114,6 +114,12 @@ def auth(
     is_flag=True,
     help="Allow tools to modify the portfolios, placing trades, etc.",
 )
+@click.option(
+    "--no-technical-tools",
+    default=False,
+    is_flag=True,
+    help="Disable optional technical analysis tools.",
+)
 @click.option(
     "--discord-token",
     type=str,
@@ -150,6 +156,7 @@ def server(
     discord_channel_id: int | None,
     discord_approver: tuple[str, ...],
     discord_timeout: int,
+    no_technical_tools: bool,
 ) -> int:
     """Run the Schwab MCP server."""
     # No logging to stderr when in MCP mode (we'll use proper MCP responses)
@@ -258,6 +265,7 @@ def server(
             client,
             approval_manager=approval_manager,
             allow_write=allow_write,
+            enable_technical_tools=not no_technical_tools,
         )
         anyio.run(server.run, backend="asyncio")
         return 0

src/schwab_mcp/server.py

@@ -55,12 +55,18 @@ def __init__(
         approval_manager: ApprovalManager,
         *,
         allow_write: bool,
+        enable_technical_tools: bool = True,
     ) -> None:
         self._server = FastMCP(
             name=name,
             lifespan=_client_lifespan(client, approval_manager),
         )
-        register_tools(self._server, client, allow_write=allow_write)
+        register_tools(
+            self._server,
+            client,
+            allow_write=allow_write,
+            enable_technical=enable_technical_tools,
+        )
 
     async def run(self) -> None:
         """Run the server using FastMCP's stdio transport."""

src/schwab_mcp/tools/__init__.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import logging
+
 from mcp.server.fastmcp import FastMCP
 from schwab.client import AsyncClient
 
@@ -9,8 +11,11 @@
 from schwab_mcp.tools import orders as _orders
 from schwab_mcp.tools import quotes as _quotes
 from schwab_mcp.tools import tools as _tools
+from schwab_mcp.tools import technical as _technical
 from schwab_mcp.tools import transactions as _txns
 
+logger = logging.getLogger(__name__)
+
 _TOOL_MODULES = (
     _tools,
     _account,
@@ -22,11 +27,21 @@
 )
 
 
-def register_tools(server: FastMCP, client: AsyncClient, *, allow_write: bool) -> None:
+def register_tools(
+    server: FastMCP,
+    client: AsyncClient,
+    *,
+    allow_write: bool,
+    enable_technical: bool = True,
+) -> None:
     """Register all Schwab tools with the provided FastMCP server."""
     _ = client
 
-    for module in _TOOL_MODULES:
+    modules = _TOOL_MODULES
+    if enable_technical:
+        modules = modules + (_technical,)
+
+    for module in modules:
         register_module = getattr(module, "register", None)
         if register_module is None:
             raise AttributeError(f"Tool module {module.__name__} missing register()")

src/schwab_mcp/tools/_registration.py

@@ -6,8 +6,10 @@
 import functools
 import inspect
 import logging
+import sys
+import types
 import uuid
-from typing import Any
+from typing import Annotated, Any, Union, cast, get_args, get_origin, get_type_hints
 
 from mcp.server.fastmcp import FastMCP, Context as MCPContext
 from mcp.types import ToolAnnotations
@@ -23,13 +25,55 @@
 _APPROVAL_WAIT_MESSAGE = "Waiting for reviewer approval…"
 
 
-def _ensure_schwab_context(func: ToolFn) -> ToolFn:
+def _is_context_annotation(annotation: Any) -> bool:
+    if annotation in (inspect._empty, None):
+        return False
+    if annotation is SchwabContext:
+        return True
+    if annotation == "SchwabContext":
+        return True
+    if isinstance(annotation, str):
+        return annotation == "SchwabContext"
+
+    origin = get_origin(annotation)
+    if origin is None:
+        return False
+
+    if origin in (Annotated,):
+        args = get_args(annotation)
+        return bool(args) and _is_context_annotation(args[0])
+
+    if origin in (Union, types.UnionType):
+        return any(_is_context_annotation(arg) for arg in get_args(annotation))
+
+    return False
+
+
+def _resolve_context_parameters(func: ToolFn) -> tuple[inspect.Signature, list[str]]:
     signature = inspect.signature(func)
-    ctx_params = [
-        name
-        for name, param in signature.parameters.items()
-        if param.annotation is SchwabContext
-    ]
+
+    module = sys.modules.get(func.__module__)
+    globalns = vars(module) if module else {}
+
+    type_hints: dict[str, Any]
+    try:
+        type_hints = get_type_hints(func, globalns=globalns, include_extras=True)
+    except TypeError:
+        type_hints = get_type_hints(func, globalns=globalns)
+    except Exception:
+        type_hints = {}
+
+    ctx_params = []
+    for name, param in signature.parameters.items():
+        annotation = type_hints.get(name, param.annotation)
+        if _is_context_annotation(annotation):
+            ctx_params.append(name)
+
+    return signature, ctx_params
+
+
+def _ensure_schwab_context(func: ToolFn) -> ToolFn:
+    signature, ctx_params = _resolve_context_parameters(func)
     if not ctx_params:
         return func
 
@@ -57,6 +101,15 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any:
             return await result
         return result
 
+    # Ensure annotations referencing names from the original module remain resolvable.
+    wrapper_globals = cast(dict[str, Any], getattr(wrapper, "__globals__", {}))
+    module = inspect.getmodule(func)
+    if module is not None:
+        module_globals = vars(module)
+        if wrapper_globals is not module_globals:
+            for key, value in module_globals.items():
+                wrapper_globals.setdefault(key, value)
+
     return wrapper
 
 
@@ -68,12 +121,7 @@ def _format_argument(value: Any) -> str:
 
 
 def _wrap_with_approval(func: ToolFn) -> ToolFn:
-    signature = inspect.signature(func)
-    ctx_params = [
-        name
-        for name, param in signature.parameters.items()
-        if param.annotation is SchwabContext
-    ]
+    signature, ctx_params = _resolve_context_parameters(func)
     if not ctx_params:
         raise TypeError(
             f"Write tool '{func.__name__}' must accept a SchwabContext parameter for approval gating."
@@ -160,6 +208,14 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any:
             raise PermissionError(message)
         raise TimeoutError(message)
 
+    wrapper_globals = cast(dict[str, Any], getattr(wrapper, "__globals__", {}))
+    module = inspect.getmodule(func)
+    if module is not None:
+        module_globals = vars(module)
+        if wrapper_globals is not module_globals:
+            for key, value in module_globals.items():
+                wrapper_globals.setdefault(key, value)
+
     return wrapper
 
 

src/schwab_mcp/tools/options.py

@@ -1,5 +1,4 @@
 #
-
 from typing import Annotated
 
 import datetime
@@ -10,6 +9,9 @@
 from schwab_mcp.tools.utils import JSONType, call
 
 
+_EXPIRATION_WINDOW_DAYS = 60
+
+
 def _parse_date(value: str | datetime.date | None) -> datetime.date | None:
     if value is None:
         return None
@@ -20,6 +22,29 @@ def _parse_date(value: str | datetime.date | None) -> datetime.date | None:
     return datetime.datetime.strptime(value, "%Y-%m-%d").date()
 
 
+def _normalize_expiration_window(
+    from_date: datetime.date | None,
+    to_date: datetime.date | None,
+    *,
+    today: datetime.date | None = None,
+) -> tuple[datetime.date | None, datetime.date | None]:
+    if from_date is None and to_date is None:
+        today = datetime.date.today() if today is None else today
+        return today, today + datetime.timedelta(days=_EXPIRATION_WINDOW_DAYS)
+
+    if from_date is None and to_date is not None:
+        today = datetime.date.today() if today is None else today
+        from_date = min(today, to_date)
+
+    if from_date is not None and to_date is None:
+        to_date = from_date + datetime.timedelta(days=_EXPIRATION_WINDOW_DAYS)
+
+    if from_date is not None and to_date is not None and to_date < from_date:
+        to_date = from_date
+
+    return from_date, to_date
+
+
 async def get_option_chain(
     ctx: SchwabContext,
     symbol: Annotated[str, "Symbol of the underlying security (e.g., 'AAPL', 'SPY')"],
@@ -45,12 +70,14 @@ async def get_option_chain(
     """
     Returns option chain data (strikes, expirations, prices) for a symbol. Use for standard chains.
     Params: symbol, contract_type (CALL/PUT/ALL), strike_count (default 25), include_quotes (bool), from_date (YYYY-MM-DD), to_date (YYYY-MM-DD).
-    Limit data returned using strike_count and date parameters.
+    Limit data returned using strike_count and date parameters. When both dates are omitted the tool defaults to the next 60 calendar days to avoid oversized responses.
     """
     client = ctx.options
 
-    from_date_obj = _parse_date(from_date)
-    to_date_obj = _parse_date(to_date)
+    from_date_obj, to_date_obj = _normalize_expiration_window(
+        _parse_date(from_date),
+        _parse_date(to_date),
+    )
 
     return await call(
         client.get_option_chain,
@@ -119,12 +146,16 @@ async def get_advanced_option_chain(
     """
     Returns advanced option chain data with strategies, filters, and theoretical calculations. Use for complex analysis.
     Params: symbol, contract_type, strike_count, include_quotes, strategy (SINGLE/ANALYTICAL/etc.), interval, strike, strike_range (ITM/NTM/etc.), from/to_date, volatility/underlying_price/interest_rate/days_to_expiration (for ANALYTICAL), exp_month, option_type (STANDARD/NON_STANDARD/ALL).
-    Limit data returned using strike_count and date parameters.
+    Limit data returned using strike_count and date parameters. When both dates are omitted the tool defaults to the next 60 calendar days to avoid oversized responses.
     """
     client = ctx.options
 
     from_date_obj = _parse_date(from_date)
     to_date_obj = _parse_date(to_date)
+    from_date_obj, to_date_obj = _normalize_expiration_window(
+        from_date_obj,
+        to_date_obj,
+    )
 
     return await call(
         client.get_option_chain,