sqlserver_mcp_catalog/services/
api_client.rs

1// SQL Server 2025 - master/msdb/sandbox combined catalog MCP server.
2//
3// Real transport: a pooled TDS connection to SQL Server (via `tiberius`),
4// replacing mcpify's originally-generated `reqwest`-based HTTP client.
5// These OpenAPI operations are synthetic (see
6// docs/sqlserver-eda-openapi-pipeline/README.md's "OpenAPI mapping
7// convention") -- there is no real HTTP endpoint on the other end, so
8// `endpoint.path`/`endpoint.method` don't describe an HTTP request; they
9// encode `/<db>/<schema>/<name>` (set by `merge_openapi.py`) and are always
10// `POST`. `endpoint.description` carries the object's `sys.objects.type_desc`
11// (`VIEW` / `SQL_STORED_PROCEDURE` / `SQL_INLINE_TABLE_VALUED_FUNCTION` /
12// `EXTENDED_STORED_PROCEDURE`), which determines the actual T-SQL shape:
13// a bare `SELECT` for views, `SELECT ... FROM name(args)` for inline
14// table-valued functions (positional args -- no named-parameter call syntax
15// exists for a FROM-clause function call), or `EXEC name @p = v, ...` for
16// stored procedures (named, so param order and optional/defaulted params
17// don't matter).
18
19use serde_json::{Map, Value};
20use tiberius::ToSql;
21
22use crate::auth::auth_manager::AuthManager;
23use crate::core::config_schema::Config;
24use crate::data::store::EndpointRecord;
25use crate::services::sql_pool;
26use crate::services::sql_type::{column_data_to_json, json_to_param};
27use crate::validation::validator::resolved_schemas_for;
28
29/// Parses `endpoint.path`'s `/<db>/<schema>/<name>` shape (set by
30/// `docs/sqlserver-eda-openapi-pipeline/tools/merge_openapi.py`) into its
31/// three parts -- the only place in this generated project that recovers
32/// which database an operation targets, since mcpify's store doesn't carry
33/// the source spec's `x-sql-database` vendor extension through.
34fn parse_path(path: &str) -> anyhow::Result<(&str, &str, &str)> {
35    let mut parts = path.trim_start_matches('/').splitn(3, '/');
36    match (parts.next(), parts.next(), parts.next()) {
37        (Some(db), Some(schema), Some(name))
38            if !db.is_empty() && !schema.is_empty() && !name.is_empty() =>
39        {
40            Ok((
41                validate_ident(db)?,
42                validate_ident(schema)?,
43                validate_ident(name)?,
44            ))
45        }
46        _ => anyhow::bail!(
47            "endpoint path '{path}' is not in the expected /<db>/<schema>/<name> shape"
48        ),
49    }
50}
51
52/// Rejects any identifier containing a character outside `[A-Za-z0-9_]` --
53/// every identifier passed through here (`db`/`schema`/`name`, stored
54/// procedure parameter names) comes from this project's own generated spec
55/// (looked up from the local SQLite store by `operation_id`, never built
56/// from a caller-supplied string), and every name in the curated object
57/// set this pipeline targets is already plain alphanumeric/underscore --
58/// see docs/sqlserver-eda-openapi-pipeline/sql/eda/allowlist.yaml. This is
59/// defense-in-depth on top of `quote_ident`'s bracket-escaping, not a
60/// response to any known injection path: T-SQL has no way to bind an
61/// object/parameter *name* as a query parameter (only values), so an
62/// identifier that ever did contain attacker-influenced text would bypass
63/// the parameterized-value protection `services::sql_type::json_to_param`
64/// provides for every argument *value*. Failing loudly here on an
65/// unexpected character is preferable to silently trusting it.
66fn validate_ident(ident: &str) -> anyhow::Result<&str> {
67    if !ident.is_empty()
68        && ident
69            .bytes()
70            .all(|b| b.is_ascii_alphanumeric() || b == b'_')
71    {
72        Ok(ident)
73    } else {
74        anyhow::bail!("identifier '{ident}' contains characters outside [A-Za-z0-9_]")
75    }
76}
77
78/// Brackets a SQL Server identifier, doubling any embedded `]` -- a second,
79/// independent layer under `validate_ident`'s charset allow-list, kept
80/// rather than dropped now that the allow-list exists, since a future
81/// change loosening the allow-list (e.g. to admit `$`/`#`-prefixed names)
82/// should not silently lose this escaping.
83fn quote_ident(ident: &str) -> String {
84    format!("[{}]", ident.replace(']', "]]"))
85}
86
87struct Param {
88    name: String,
89    ordinal: u64,
90    x_sql_type: String,
91}
92
93/// Reads `properties`/`x-sql-ordinal`/`x-sql-type` off a resolved input
94/// schema (see `tools/generate_openapi.py`'s `build_request_schema`),
95/// sorted by declared ordinal -- required for a table-valued function's
96/// positional call syntax, and used for stored procedures' named syntax
97/// mainly for a deterministic/readable generated statement.
98fn ordered_params(input_schema: &Value) -> Vec<Param> {
99    let mut params: Vec<Param> = input_schema
100        .get("properties")
101        .and_then(Value::as_object)
102        .into_iter()
103        .flatten()
104        .map(|(name, schema)| Param {
105            name: name.clone(),
106            ordinal: schema
107                .get("x-sql-ordinal")
108                .and_then(Value::as_u64)
109                .unwrap_or(0),
110            x_sql_type: schema
111                .get("x-sql-type")
112                .and_then(Value::as_str)
113                .unwrap_or("nvarchar(max)")
114                .to_string(),
115        })
116        .collect();
117    params.sort_by_key(|p| p.ordinal);
118    params
119}
120
121/// Builds the T-SQL text (with `@P1`/`@P2`/... placeholders) and the
122/// matching positionally-ordered bound parameters for one operation call.
123fn build_statement(
124    db: &str,
125    schema: &str,
126    name: &str,
127    kind: Option<&str>,
128    params: &[Param],
129    body: &Map<String, Value>,
130) -> anyhow::Result<(String, Vec<Box<dyn ToSql>>)> {
131    let qualified = format!(
132        "{}.{}.{}",
133        quote_ident(db),
134        quote_ident(schema),
135        quote_ident(name)
136    );
137
138    let mut bound: Vec<Box<dyn ToSql>> = Vec::with_capacity(params.len());
139    for param in params {
140        let value = body.get(&param.name).cloned().unwrap_or(Value::Null);
141        bound.push(json_to_param(&value, &param.x_sql_type)?);
142    }
143
144    let sql = match kind {
145        Some("VIEW") => format!("SELECT * FROM {qualified}"),
146        Some(k) if k.ends_with("_FUNCTION") || k.contains("TABLE_VALUED_FUNCTION") => {
147            let placeholders = (1..=bound.len())
148                .map(|i| format!("@P{i}"))
149                .collect::<Vec<_>>()
150                .join(", ");
151            format!("SELECT * FROM {qualified}({placeholders})")
152        }
153        // Stored procedures (SQL_STORED_PROCEDURE/EXTENDED_STORED_PROCEDURE)
154        // and the `kind.is_none()` fallback (an older store built before
155        // `description` carried this classification) both take the EXEC
156        // path -- named parameter syntax, so a proc with only optional
157        // parameters and no arguments supplied still works (`bound` empty
158        // produces a bare `EXEC name`).
159        _ => {
160            if bound.is_empty() {
161                format!("EXEC {qualified}")
162            } else {
163                let mut assignments = Vec::with_capacity(params.len());
164                for (i, p) in params.iter().enumerate() {
165                    assignments.push(format!(
166                        "{} = @P{}",
167                        quote_ident(validate_ident(&p.name)?),
168                        i + 1
169                    ));
170                }
171                format!("EXEC {qualified} {}", assignments.join(", "))
172            }
173        }
174    };
175
176    Ok((sql, bound))
177}
178
179/// Maps a `tiberius` error to this project's synthetic 400/403/500 split
180/// (see `docs/sqlserver-eda-openapi-pipeline/README.md`'s "OpenAPI mapping
181/// convention" -- SQL Server severity 11-16 statement errors are `400`,
182/// severity-14 permission errors are `403`, severity 17-25 engine/fatal
183/// errors are `500`), rather than surfacing the raw driver error
184/// undifferentiated. Returned as a JSON value matching
185/// `components.schemas.SqlServerError` shape rather than a bare string, so
186/// callers get the same structured error either way they might have gotten
187/// it from a real synchronous SQL Server error.
188fn classify_tiberius_error(err: tiberius::error::Error) -> anyhow::Error {
189    use tiberius::error::Error as TError;
190    match &err {
191        TError::Server(token) => {
192            let (number, state, class, message, procedure, line) = (
193                token.code(),
194                token.state(),
195                token.class(),
196                token.message(),
197                token.procedure(),
198                token.line(),
199            );
200            let category = if class >= 17 {
201                "fatal/resource error (500-equivalent)"
202            } else if class == 14 {
203                "permission denied (403-equivalent)"
204            } else {
205                "statement/user error (400-equivalent)"
206            };
207            anyhow::anyhow!(
208                "SQL Server error {number} (severity {class}, state {state}) in '{procedure}' line {line}: {message} [{category}]"
209            )
210        }
211        other => anyhow::anyhow!("SQL Server connection/protocol error (500-equivalent): {other}"),
212    }
213}
214
215pub struct ApiClient {
216    config: Config,
217}
218
219impl ApiClient {
220    pub fn new(config: Config) -> Self {
221        Self { config }
222    }
223
224    /// Executes `endpoint` against the configured SQL Server instance.
225    /// `auth_manager` resolves to a `tiberius::AuthMethod` (see
226    /// `auth::auth_manager::AuthManager::resolve_tds_auth`) from the
227    /// server's own configured credentials -- there is no per-request
228    /// credential override; SQL Server auth is always this server's own
229    /// configured identity, not something a caller supplies per call.
230    pub async fn execute(
231        &self,
232        endpoint: &EndpointRecord,
233        args: &Value,
234        auth_manager: &mut AuthManager,
235    ) -> anyhow::Result<Value> {
236        let (db, schema, name) = parse_path(&endpoint.path)?;
237
238        let (input_schema, _output_schema) =
239            resolved_schemas_for(&self.config.api_version, &endpoint.operation_id).ok_or_else(
240                || {
241                    anyhow::anyhow!(
242                        "no resolved schema found for operation '{}' under api_version '{}'",
243                        endpoint.operation_id,
244                        self.config.api_version
245                    )
246                },
247            )?;
248        let params = ordered_params(input_schema);
249
250        let empty = Map::new();
251        let args_map = args.as_object().unwrap_or(&empty);
252        let body = args_map
253            .get("body")
254            .and_then(Value::as_object)
255            .cloned()
256            .unwrap_or_default();
257
258        let (sql, bound) = build_statement(
259            db,
260            schema,
261            name,
262            endpoint.description.as_deref(),
263            &params,
264            &body,
265        )?;
266
267        let auth_method = auth_manager.resolve_tds_auth().await?;
268
269        let (host, port) = self.config.host_and_port();
270        let mut tiberius_config = tiberius::Config::new();
271        tiberius_config.host(host);
272        tiberius_config.port(port);
273        tiberius_config.authentication(auth_method);
274        if self.config.trust_server_cert {
275            // Curated system-catalog objects, not user data -- trusting the
276            // server cert matches this project's primary use case (a local
277            // Docker/dev instance with a self-signed cert; see
278            // docs/sqlserver-eda-openapi-pipeline/docker-compose.yml)
279            // rather than requiring every operator to supply a CA bundle
280            // before the first query works. Set `trust_server_cert: false`
281            // for a production instance with a real CA-signed cert.
282            tiberius_config.trust_cert();
283        }
284
285        let pool_key = format!("{host}:{port}");
286        let pool =
287            sql_pool::cached_pool(&pool_key, tiberius_config, self.config.pool_max_size).await?;
288        let mut conn = pool.get().await.map_err(|err| {
289            anyhow::anyhow!("failed to obtain a pooled SQL Server connection: {err}")
290        })?;
291
292        let bound_refs: Vec<&dyn ToSql> = bound.iter().map(|param| param.as_ref()).collect();
293        let stream = conn
294            .query(&sql, &bound_refs)
295            .await
296            .map_err(classify_tiberius_error)?;
297        let rows = stream
298            .into_first_result()
299            .await
300            .map_err(classify_tiberius_error)?;
301
302        let json_rows: Vec<Value> = rows
303            .iter()
304            .map(|row| {
305                let mut obj = Map::new();
306                for (column, data) in row.cells() {
307                    obj.insert(column.name().to_string(), column_data_to_json(data));
308                }
309                Value::Object(obj)
310            })
311            .collect();
312        Ok(Value::Array(json_rows))
313    }
314}
315
316#[cfg(test)]
317mod tests {
318    use super::*;
319
320    #[test]
321    fn parse_path_splits_db_schema_name() {
322        assert_eq!(
323            parse_path("/master/INFORMATION_SCHEMA/COLUMNS").unwrap(),
324            ("master", "INFORMATION_SCHEMA", "COLUMNS")
325        );
326    }
327
328    #[test]
329    fn parse_path_rejects_a_path_with_too_few_segments() {
330        assert!(parse_path("/master/COLUMNS").is_err());
331    }
332
333    #[test]
334    fn quote_ident_doubles_embedded_closing_brackets() {
335        assert_eq!(quote_ident("weird]name"), "[weird]]name]");
336    }
337
338    #[test]
339    fn validate_ident_accepts_alphanumeric_and_underscore() {
340        assert!(validate_ident("sp_who2").is_ok());
341        assert!(validate_ident("INFORMATION_SCHEMA").is_ok());
342    }
343
344    #[test]
345    fn validate_ident_rejects_anything_else() {
346        assert!(validate_ident("").is_err());
347        assert!(validate_ident("robert'; drop table endpoints;--").is_err());
348        assert!(validate_ident("weird]name").is_err());
349        assert!(validate_ident("has space").is_err());
350    }
351
352    #[test]
353    fn parse_path_rejects_a_path_with_an_unsafe_segment() {
354        assert!(parse_path("/master/sys/sp_who; DROP TABLE endpoints--").is_err());
355    }
356
357    #[test]
358    fn build_statement_selects_from_a_view_with_no_parameters() {
359        let (sql, bound) = build_statement(
360            "master",
361            "INFORMATION_SCHEMA",
362            "COLUMNS",
363            Some("VIEW"),
364            &[],
365            &Map::new(),
366        )
367        .unwrap();
368        assert_eq!(sql, "SELECT * FROM [master].[INFORMATION_SCHEMA].[COLUMNS]");
369        assert!(bound.is_empty());
370    }
371
372    #[test]
373    fn build_statement_execs_a_proc_with_named_parameters() {
374        let params = vec![
375            Param {
376                name: "objname".to_string(),
377                ordinal: 1,
378                x_sql_type: "nvarchar(1035)".to_string(),
379            },
380            Param {
381                name: "newname".to_string(),
382                ordinal: 2,
383                x_sql_type: "sysname".to_string(),
384            },
385        ];
386        let mut body = Map::new();
387        body.insert("objname".to_string(), Value::String("t1".to_string()));
388        body.insert("newname".to_string(), Value::String("t2".to_string()));
389        let (sql, bound) = build_statement(
390            "master",
391            "sys",
392            "sp_rename",
393            Some("SQL_STORED_PROCEDURE"),
394            &params,
395            &body,
396        )
397        .unwrap();
398        assert_eq!(
399            sql,
400            "EXEC [master].[sys].[sp_rename] [objname] = @P1, [newname] = @P2"
401        );
402        assert_eq!(bound.len(), 2);
403    }
404
405    #[test]
406    fn build_statement_selects_from_a_function_with_positional_placeholders() {
407        let params = vec![Param {
408            name: "session_id".to_string(),
409            ordinal: 1,
410            x_sql_type: "smallint".to_string(),
411        }];
412        let mut body = Map::new();
413        body.insert("session_id".to_string(), Value::from(52));
414        let (sql, bound) = build_statement(
415            "master",
416            "sys",
417            "dm_exec_sql_text",
418            Some("SQL_INLINE_TABLE_VALUED_FUNCTION"),
419            &params,
420            &body,
421        )
422        .unwrap();
423        assert_eq!(sql, "SELECT * FROM [master].[sys].[dm_exec_sql_text](@P1)");
424        assert_eq!(bound.len(), 1);
425    }
426
427    #[test]
428    fn build_statement_execs_a_parameterless_proc() {
429        let (sql, bound) = build_statement(
430            "master",
431            "sys",
432            "sp_who",
433            Some("SQL_STORED_PROCEDURE"),
434            &[],
435            &Map::new(),
436        )
437        .unwrap();
438        assert_eq!(sql, "EXEC [master].[sys].[sp_who]");
439        assert!(bound.is_empty());
440    }
441}
sqlserver_mcp_catalog/services/api_client.rs

sqlserver_mcp_catalog/services/
api_client.rs
