Python: Add models for the MCP server SDKs (mcp, fastmcp)

python/ql/lib/change-notes/2026-09-30-mcp-models.md

@@ -0,0 +1,4 @@
+---
+category: minorAnalysis
+---
+* Added modeling of the `mcp` (the official Model Context Protocol SDK) and `fastmcp` PyPI packages. The parameters of MCP tool, resource and prompt handlers are now sources of the opt-in `ai` threat model, as their values are produced by a language model; the `Context` parameter that the SDKs inject is not a source. The HTTP headers that `fastmcp` handlers read through `get_http_headers()` or `get_http_request()`, and the bearer token that token verifiers (`verify_token`) and `get_access_token()` receive, are sources of the `remote` threat model.

python/ql/lib/semmle/python/Frameworks.qll

@@ -50,6 +50,7 @@ private import semmle.python.frameworks.Libxml2
 private import semmle.python.frameworks.Lxml
 private import semmle.python.frameworks.Mako
 private import semmle.python.frameworks.MarkupSafe
+private import semmle.python.frameworks.Mcp
 private import semmle.python.frameworks.Multidict
 private import semmle.python.frameworks.Mysql
 private import semmle.python.frameworks.MySQLdb

python/ql/lib/semmle/python/frameworks/Mcp.qll

@@ -0,0 +1,271 @@
+/**
+ * Provides classes modeling security-relevant aspects of the `mcp` and `fastmcp` PyPI packages,
+ * which implement Model Context Protocol (MCP) servers.
+ *
+ * See
+ * - https://pypi.org/project/mcp/
+ * - https://pypi.org/project/fastmcp/
+ */
+
+private import python
+private import semmle.python.Concepts
+private import semmle.python.ApiGraphs
+
+/**
+ * INTERNAL: Do not use.
+ *
+ * Provides models for the `mcp` and `fastmcp` PyPI packages.
+ *
+ * The parameters of tool, resource and prompt handlers are chosen by the MCP client, that is by a
+ * language model or by anyone who can reach the server. The low-level server of `mcp`, the HTTP
+ * headers and the bearer tokens are modeled as data extensions (`mcp.model.yml`, `fastmcp.model.yml`).
+ */
+module Mcp {
+  /** Gets a reference to a server class whose instances register tool, resource and prompt handlers. */
+  private API::Node serverClass() {
+    exists(API::Node mcpServer | mcpServer = API::moduleImport("mcp").getMember("server") |
+      // `FastMCP` in 1.x of `mcp`, renamed to `MCPServer` in 2.x
+      result = mcpServer.getMember("FastMCP")
+      or
+      result = mcpServer.getMember("fastmcp").getMember("FastMCP")
+      or
+      result = mcpServer.getMember("fastmcp").getMember("server").getMember("FastMCP")
+      or
+      result = mcpServer.getMember("MCPServer")
+      or
+      result = mcpServer.getMember("mcpserver").getMember("MCPServer")
+      or
+      result = mcpServer.getMember("mcpserver").getMember("server").getMember("MCPServer")
+    )
+    or
+    exists(API::Node fastmcp | fastmcp = API::moduleImport("fastmcp") |
+      result = fastmcp.getMember("FastMCP")
+      or
+      result = fastmcp.getMember("server").getMember("FastMCP")
+      or
+      result = fastmcp.getMember("server").getMember("server").getMember("FastMCP")
+    )
+  }
+
+  /** Gets a reference to a tool or prompt class of `fastmcp` that has a `from_function` method. */
+  private API::Node functionComponentClass() {
+    exists(API::Node tools | tools = API::moduleImport("fastmcp").getMember("tools") |
+      result = tools.getMember(["Tool", "FunctionTool"])
+      or
+      // 2.x defines both classes in `fastmcp.tools.tool`, later versions split them up
+      result = tools.getMember("tool").getMember(["Tool", "FunctionTool"])
+      or
+      result = tools.getMember("base").getMember("Tool")
+      or
+      result = tools.getMember("function_tool").getMember("FunctionTool")
+    )
+    or
+    exists(API::Node prompts | prompts = API::moduleImport("fastmcp").getMember("prompts") |
+      result = prompts.getMember(["Prompt", "FunctionPrompt"])
+      or
+      result = prompts.getMember("prompt").getMember(["Prompt", "FunctionPrompt"])
+      or
+      result = prompts.getMember("base").getMember("Prompt")
+      or
+      result = prompts.getMember("function_prompt").getMember("FunctionPrompt")
+    )
+  }
+
+  /**
+   * Gets a reference to a standalone decorator of `fastmcp` (3.x and later) that marks a function
+   * as a tool or prompt.
+   */
+  private API::Node standaloneDecorator() {
+    exists(API::Node fastmcp | fastmcp = API::moduleImport("fastmcp") |
+      result = fastmcp.getMember("tools").getMember("tool")
+      or
+      result = fastmcp.getMember("tools").getMember("function_tool").getMember("tool")
+      or
+      result = fastmcp.getMember("prompts").getMember("prompt")
+      or
+      result = fastmcp.getMember("prompts").getMember("function_prompt").getMember("prompt")
+    )
+  }
+
+  /** Gets an API node for a function that is registered as a tool, resource or prompt handler. */
+  private API::Node handlerNode() {
+    exists(API::Node server | server = serverClass().getAnInstance() |
+      // `@server.tool()`, `@server.resource(uri)`, `@server.prompt()`
+      result = server.getMember(["tool", "resource", "prompt"]).getReturn().getParameter(0)
+      or
+      // `@server.tool` and `@server.prompt` without parentheses (`fastmcp`)
+      result = server.getMember(["tool", "prompt"]).getParameter(0)
+      or
+      result = server.getMember(["tool", "prompt"]).getKeywordParameter("name_or_fn")
+      or
+      result = server.getMember(["add_tool", "add_prompt"]).getParameter(0)
+      or
+      result = server.getMember("add_tool").getKeywordParameter(["fn", "tool"])
+      or
+      result = server.getMember("add_prompt").getKeywordParameter("prompt")
+    )
+    or
+    exists(API::Node fromFunction |
+      fromFunction = functionComponentClass().getMember("from_function")
+    |
+      result = fromFunction.getParameter(0)
+      or
+      result = fromFunction.getKeywordParameter("fn")
+    )
+    or
+    exists(API::Node decorator | decorator = standaloneDecorator() |
+      result = decorator.getParameter(0)
+      or
+      result = decorator.getKeywordParameter("name_or_fn")
+      or
+      result = decorator.getReturn().getParameter(0)
+    )
+  }
+
+  /** Gets a reference to the `Context` class whose instances the SDKs inject into handlers. */
+  private API::Node contextClass() {
+    exists(API::Node mcpServer | mcpServer = API::moduleImport("mcp").getMember("server") |
+      result = mcpServer.getMember("fastmcp").getMember("Context")
+      or
+      result = mcpServer.getMember("fastmcp").getMember("server").getMember("Context")
+      or
+      result = mcpServer.getMember("mcpserver").getMember("Context")
+      or
+      result = mcpServer.getMember("mcpserver").getMember("context").getMember("Context")
+    )
+    or
+    exists(API::Node fastmcp | fastmcp = API::moduleImport("fastmcp") |
+      result = fastmcp.getMember("Context")
+      or
+      result = fastmcp.getMember("server").getMember("Context")
+      or
+      result = fastmcp.getMember("server").getMember("context").getMember("Context")
+    )
+  }
+
+  /** Gets a reference to the `Context` class or to a subclass of it; the SDKs inject both. */
+  private API::Node contextClassOrSubclass() { result = contextClass().getASubclass*() }
+
+  /** Holds if `def` binds a module-level name to the `Context` class or to a subclass of it. */
+  private predicate bindsContext(Name def) {
+    def.getScope() instanceof Module and
+    (
+      // `class ProjectContext(Context): ...`
+      def = contextClassOrSubclass().getAValueReachableFromSource().asExpr()
+      or
+      // `from fastmcp import Context as Ctx`
+      exists(Alias alias | alias.getAsname() = def |
+        alias.getValue() = contextClassOrSubclass().getAValueReachableFromSource().asExpr()
+      )
+    )
+  }
+
+  /**
+   * Gets a name that refers to the `Context` class or to a subclass of it in the module `m`. The SDKs
+   * resolve string annotations through the globals of the module, so a name that `m` also binds to
+   * something else is left out.
+   */
+  private string contextName(Module m) {
+    exists(Name def | bindsContext(def) and def.getScope() = m and result = def.getId()) and
+    not exists(Name other |
+      other.getScope() = m and
+      other.isDefinition() and
+      other.getId() = result and
+      not bindsContext(other)
+    )
+  }
+
+  /** Holds if `s` subscripts the `typing` (or `typing_extensions`) member `name`. */
+  private predicate subscriptsTyping(Subscript s, string name) {
+    s.getObject() =
+      API::moduleImport(["typing", "typing_extensions"])
+          .getMember(name)
+          .getAValueReachableFromSource()
+          .asExpr()
+  }
+
+  /**
+   * Holds if the string annotation `s` refers to the `Context` class in a type position, as in
+   * `"Context"`, `"Context[...]"`, `"Context | None"`, `"Optional[Context]"`,
+   * `"Union[Context, None]"` or `"Annotated[Context, ...]"`. Strings of any other form, among them
+   * those with `Context` only in metadata or in `Literal` values, do not count.
+   */
+  private predicate isContextString(StringLiteral s) {
+    exists(string context, string other |
+      // the class, possibly with type arguments
+      context = contextName(s.getEnclosingModule()) + "(\\[.*\\])?" and
+      // a simple type in a union, such as `None`, `str` or `typing.Any`
+      other = "[\\w.]+(\\[[^\\]]*\\])?"
+    |
+      s.getText()
+          .regexpMatch("\\s*(" + other + "\\s*\\|\\s*)*" + context + "(\\s*\\|\\s*" + other +
+              ")*\\s*")
+      or
+      s.getText().regexpMatch("\\s*([\\w]+\\.)*Optional\\[\\s*" + context + "\\s*\\]\\s*")
+      or
+      s.getText()
+          .regexpMatch("\\s*([\\w]+\\.)*Union\\[\\s*(" + other + "\\s*,\\s*)*" + context +
+              "(\\s*,\\s*" + other + ")*\\s*\\]\\s*")
+      or
+      s.getText().regexpMatch("\\s*([\\w]+\\.)*Annotated\\[\\s*" + context + "\\s*,.*\\]\\s*")
+    )
+  }
+
+  /**
+   * Holds if the annotation `e` refers to the injected `Context` class, possibly as
+   * `Context[...]`, `Context | None`, `Optional[Context]`, `Union[Context, ...]`,
+   * `Annotated[Context, ...]` or as a string such as `"Context"`.
+   */
+  private predicate isContextAnnotation(Expr e) {
+    e = contextClassOrSubclass().getAValueReachableFromSource().asExpr()
+    or
+    isContextAnnotation(e.(Subscript).getObject())
+    or
+    // Only the first argument of `Annotated` is a type, the others are metadata.
+    exists(Subscript s | s = e and subscriptsTyping(s, "Annotated") |
+      isContextAnnotation(s.getIndex().(Tuple).getElt(0))
+    )
+    or
+    // The arguments of `Literal` are values, not types.
+    exists(Subscript s | s = e and not subscriptsTyping(s, ["Annotated", "Literal"]) |
+      isContextAnnotation(s.getIndex())
+    )
+    or
+    isContextAnnotation(e.(Tuple).getAnElt())
+    or
+    exists(BinaryExpr union | union = e and union.getOp() instanceof BitOr |
+      isContextAnnotation(union.getLeft()) or isContextAnnotation(union.getRight())
+    )
+    or
+    isContextString(e)
+  }
+
+  /**
+   * Holds if `p` is a parameter of `handler` whose value is chosen by the MCP client.
+   *
+   * A handler decorated in a class body is registered unbound, so the client supplies its first
+   * parameter as well.
+   */
+  private predicate isClientParameter(Function handler, Parameter p) {
+    handler = handlerNode().getAValueReachingSink().asExpr().(CallableExpr).getInnerScope() and
+    (
+      p = handler.getAnArg()
+      or
+      p = handler.getAKeywordOnlyArg()
+      or
+      p = handler.getVararg()
+      or
+      p = handler.getKwarg()
+    ) and
+    not isContextAnnotation(p.getAnnotation())
+  }
+
+  /** A parameter of an MCP tool, resource or prompt handler, seen as a source of the `ai` threat model. */
+  private class HandlerParameter extends ThreatModelSource::Range, DataFlow::ParameterNode {
+    HandlerParameter() { isClientParameter(_, this.getParameter()) }
+
+    override string getThreatModel() { result = "ai" }
+
+    override string getSourceType() { result = "MCP handler parameter" }
+  }
+}

python/ql/lib/semmle/python/frameworks/fastmcp.model.yml

@@ -0,0 +1,22 @@
+extensions:
+  - addsTo:
+      pack: codeql/python-all
+      extensible: sourceModel
+    data:
+      # `fastmcp` (PyPI), a standalone MCP server framework. The parameters of tool, resource and
+      # prompt handlers are modeled in `Mcp.qll`, which leaves out the injected `Context` parameter.
+      # The HTTP headers of the transport.
+      - ['fastmcp', 'Member[server].Member[dependencies].Member[get_http_headers].ReturnValue', 'remote']
+      - ['fastmcp', 'Member[server].Member[dependencies].Member[get_http_request].ReturnValue.Member[headers]', 'remote']
+      # The bearer token of the Authorization header reaches token verifiers and the access-token helper.
+      - ['fastmcp.server.auth.TokenVerifier!', 'Subclass.Member[verify_token].Parameter[0,token:]', 'remote']
+      - ['fastmcp.server.auth.AuthProvider!', 'Subclass.Member[verify_token].Parameter[0,token:]', 'remote']
+      - ['fastmcp', 'Member[server].Member[dependencies].Member[get_access_token].ReturnValue.Member[token]', 'remote']
+
+  - addsTo:
+      pack: codeql/python-all
+      extensible: typeModel
+    data:
+      # The classes are re-exported; instances of any import path are instances of the class.
+      - ['fastmcp.server.auth.TokenVerifier!', 'fastmcp', 'Member[server].Member[auth].Member[auth].Member[TokenVerifier]']
+      - ['fastmcp.server.auth.AuthProvider!', 'fastmcp', 'Member[server].Member[auth].Member[auth].Member[AuthProvider]']

python/ql/lib/semmle/python/frameworks/mcp.model.yml

@@ -0,0 +1,24 @@
+extensions:
+  - addsTo:
+      pack: codeql/python-all
+      extensible: sourceModel
+    data:
+      # `mcp`, the official Model Context Protocol SDK. The handlers of `FastMCP` (1.x) and `MCPServer`
+      # (2.x) are modeled in `Mcp.qll`, which leaves out the `Context` parameter that the SDK injects.
+      # The low-level server: decorators in 1.x, constructor handlers taking `(ctx, params)` in 2.x.
+      # Its handlers take no `Context`; their parameters are chosen by the MCP client.
+      - ['mcp.server.lowlevel.Server', 'Member[call_tool,read_resource,get_prompt].ReturnValue.Argument[0].Parameter[any]', 'ai']
+      - ['mcp.server.lowlevel.Server!', 'Argument[on_call_tool:,on_get_prompt:].Parameter[1].Member[arguments,name]', 'ai']
+      - ['mcp.server.lowlevel.Server!', 'Argument[on_read_resource:].Parameter[1].Member[uri]', 'ai']
+      # The bearer token of the Authorization header reaches token verifiers and the access-token helper.
+      - ['mcp.server.auth.provider.TokenVerifier!', 'Subclass.Member[verify_token].Parameter[0,token:]', 'remote']
+      - ['mcp', 'Member[server].Member[auth].Member[middleware].Member[auth_context].Member[get_access_token].ReturnValue.Member[token]', 'remote']
+
+  - addsTo:
+      pack: codeql/python-all
+      extensible: typeModel
+    data:
+      # The class is re-exported; instances of any import path are instances of the class.
+      - ['mcp.server.lowlevel.Server!', 'mcp', 'Member[server].Member[Server]']
+      - ['mcp.server.lowlevel.Server!', 'mcp', 'Member[server].Member[lowlevel].Member[server].Member[Server]']
+      - ['mcp.server.lowlevel.Server', 'mcp.server.lowlevel.Server!', 'Instance']

python/ql/test/library-tests/frameworks/fastmcp/ConceptsTest.ql

@@ -0,0 +1,2 @@
+import python
+import experimental.meta.ConceptsTest

python/ql/test/library-tests/frameworks/fastmcp/MaDTest.ql

@@ -0,0 +1,2 @@
+import python
+import experimental.meta.MaDTest