SQL Server MCP

Instructions

Args:
    procedure: Procedure name, optionally with schema (e.g., 'dbo.sp_GetUser' or 'sp_GetUser')
    params: Input parameter values as dictionary (parameter names without @)

Returns:
    Dictionary with:
    - procedure: Full procedure name
    - result_sets: List of result sets (each is a list of row dictionaries)
    - status: 'success' or error

Implementation Reference

• src/mssql_mcp/tools/stored_procs.py:143-198 (handler)

  The primary MCP tool handler for 'call_stored_proc'. Decorated with @mcp.tool(), it parses the procedure name, handles parameters, checks read-only mode, and calls the low-level ConnectionManager method.

  @mcp.tool()
  def call_stored_proc(
      procedure: str,
      params: dict[str, Any] | None = None,
  ) -> dict[str, Any]:
      """Execute a stored procedure.
  
      Args:
          procedure: Procedure name, optionally with schema (e.g., 'dbo.sp_GetUser' or 'sp_GetUser')
          params: Input parameter values as dictionary (parameter names without @)
  
      Returns:
          Dictionary with:
          - procedure: Full procedure name
          - result_sets: List of result sets (each is a list of row dictionaries)
          - status: 'success' or error
      """
      try:
          manager = get_connection_manager()
          config = manager.config
  
          # Check read-only mode
          if config.read_only:
              return {"error": "Stored procedure execution disabled in read-only mode"}
  
          # Parse schema.procedure format
          if "." in procedure:
              parts = procedure.split(".", 1)
              schema = parts[0]
              proc_name = parts[1]
          else:
              schema = "dbo"
              proc_name = procedure
  
          full_name = f"{schema}.{proc_name}"
  
          # Build parameter tuple
          param_values = tuple(params.values()) if params else None
  
          # Execute stored procedure
          results = manager.call_stored_proc(full_name, param_values)
  
          return {
              "status": "success",
              "procedure": full_name,
              "result_sets": [results] if results else [],
              "row_count": len(results) if results else 0,
          }
  
      except QueryError as e:
          logger.error(f"Error calling stored procedure {procedure}: {e}")
          return {"error": str(e)}
      except Exception as e:
          logger.error(f"Unexpected error calling stored procedure {procedure}: {e}")
          return {"error": str(e)}

• src/mssql_mcp/connection.py:252-287 (helper)

  Low-level helper method in ConnectionManager that performs the actual stored procedure execution using pymssql's cursor.callproc.

  def call_stored_proc(
      self,
      proc_name: str,
      params: tuple[Any, ...] | None = None,
  ) -> list[dict[str, Any]]:
      """Execute a stored procedure and return results.
  
      Args:
          proc_name: Name of the stored procedure
          params: Optional procedure parameters
  
      Returns:
          List of result rows as dicts
  
      Raises:
          QueryError: If execution fails
      """
      conn = self.get_connection()
      try:
          cursor = conn.cursor(as_dict=True)
          cursor.callproc(proc_name, params or ())
  
          # Fetch results if any
          results = []
          if cursor.description:
              results = list(cursor.fetchall())
  
          conn.commit()
          return results
  
      except pymssql.Error as e:
          logger.error(f"Stored procedure execution failed: {e}")
          raise QueryError(f"Stored procedure '{proc_name}' failed: {e}") from e
      finally:
          cursor.close()