salesforce_query
Execute SOQL queries to retrieve and analyze Salesforce data in JSON format, enabling direct access to organization records and information.
Instructions
Run a SOQL query and return JSON rows
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| args | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| records | Yes | ||
| total_size | Yes |
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")