Salesforce MCP Server

Instructions

Implementation Reference

• sfmcp/tools/query.py:23-28 (handler)

  The core asynchronous handler function for the 'salesforce_query' MCP tool. It uses SalesforceClient to run the SOQL query and returns results as QueryResult.

  async def salesforce_query(args: QueryArgs) -> QueryResult:
      sf = SalesforceClient.from_env()
      rows = await sf.run_soql(args.soql)
      if args.max_records is not None:
          rows = rows[: args.max_records]
      return QueryResult(total_size=len(rows), records=rows)

• sfmcp/tools/query.py:9-16 (schema)

  Pydantic models for input (QueryArgs: soql and optional max_records) and output (QueryResult: total_size and records list) schema validation.

  class QueryArgs(BaseModel):
      soql: str = Field(..., description="SOQL query string")
      max_records: int | None = Field(None, ge=1, le=50000)
  
  
  class QueryResult(BaseModel):
      total_size: int
      records: List[Dict[str, Any]]

• sfmcp/tools/query.py:19-29 (registration)

  Module register function using FastMCP @tool decorator to define and register the salesforce_query tool.

  def register(mcp: FastMCP) -> None:
      @mcp.tool(
          name="salesforce_query", description="Run a SOQL query and return JSON rows"
      )
      async def salesforce_query(args: QueryArgs) -> QueryResult:
          sf = SalesforceClient.from_env()
          rows = await sf.run_soql(args.soql)
          if args.max_records is not None:
              rows = rows[: args.max_records]
          return QueryResult(total_size=len(rows), records=rows)

• sfmcp/server.py:23-33 (registration)

  Top-level registration in server.py: imports tool_query from .tools.query and calls tool_query.register(mcp) to activate the salesforce_query tool among others.

  def _register_all() -> None:
      tool_query.register(mcp)
      tool_describe.register(mcp)
      tool_list_objects.register(mcp)
      tool_list_flows.register(mcp)
      tool_list_reports.register(mcp)
      tool_list_dashboards.register(mcp)
      tool_describe_flow.register(mcp)
      # res_saved_queries.register(mcp)
      # prm_opps_by_stage.register(mcp)

• sfmcp/salesforce_client.py:58-77 (helper)

  Key helper method in SalesforceClient that executes the SOQL query using Salesforce CLI subprocess and parses the JSON response to return records.

  async def run_soql(self, soql: str) -> List[Dict[str, Any]]:
      """Run a SOQL query and return the records"""
      command = [
          "sf",
          "data",
          "query",
          "--target-org",
          self._org_alias,
          "--query",
          soql,
          "--result-format",
          "json",
      ]
      result = await self._run_cli_command(command)
  
      if "result" in result and "records" in result["result"]:
          return result["result"]["records"]  # type: ignore[no-any-return]
      else:
          raise Exception("Unexpected response format from Salesforce CLI")