Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[project]
name = "sap-cloud-sdk"
version = "0.57.1"
version = "0.57.2"
description = "SAP Cloud SDK for Python"
readme = "README.md"
license = "Apache-2.0"
Expand Down
55 changes: 53 additions & 2 deletions src/sap_cloud_sdk/agentgateway/converters.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,44 @@
"object": dict,
}

# JSON Schema keys that map directly to a Pydantic Field kwarg (same semantics,
# but camelCase → snake_case where needed).
_FIELD_KWARGS: dict[str, str] = {
"title": "title",
"description": "description",
"examples": "examples",
"deprecated": "deprecated",
"pattern": "pattern",
"minLength": "min_length",
"maxLength": "max_length",
"minimum": "ge",
"maximum": "le",
"exclusiveMinimum": "gt",
"exclusiveMaximum": "lt",
"multipleOf": "multiple_of",
}

# Everything else the MCP builder can emit that has no native Pydantic Field kwarg.
# These are passed through via json_schema_extra so the LLM still sees them.
_EXTRA_KEYS: frozenset[str] = frozenset(
{
"enum",
"default",
"example",
"const",
"format",
"contentEncoding",
"uniqueItems",
"items",
"properties",
"required",
"additionalProperties",
"oneOf",
"anyOf",
"allOf",
}
)


def _resolve_type(json_type: Any) -> tuple[type, bool]:
"""Return (python_type, is_nullable) from a JSON Schema ``type`` value.
Expand Down Expand Up @@ -119,10 +157,23 @@ async def run(**kwargs) -> str:
v = properties[orig]
py_type, type_nullable = _resolve_type(v.get("type"))
optional = orig not in required

field_kwargs: dict[str, Any] = {}
extra: dict[str, Any] = {}
for key, value in v.items():
if key in ("type",):
continue
if key in _FIELD_KWARGS:
field_kwargs[_FIELD_KWARGS[key]] = value
elif key in _EXTRA_KEYS:
extra[key] = value
if extra:
field_kwargs["json_schema_extra"] = extra

if optional or type_nullable:
fields[safe] = (py_type | None, Field(default=None))
fields[safe] = (py_type | None, Field(default=None, **field_kwargs))
else:
fields[safe] = (py_type, ...)
fields[safe] = (py_type, Field(..., **field_kwargs))
args_schema = create_model(f"{mcp_tool.name}_args", **fields) if fields else None

return StructuredTool.from_function(
Expand Down
8 changes: 8 additions & 0 deletions tests/agentgateway/integration/agw_auth.feature
Original file line number Diff line number Diff line change
Expand Up @@ -59,3 +59,11 @@ Feature: Agent Gateway Auth Integration
Scenario: Get IAS client ID returns a non-empty string
When I call get_ias_client_id
Then the ias_client_id should be a non-empty string

Scenario: Convert MCP tools to LangChain StructuredTools
Given I have a valid user token
When I call list_mcp_tools
And I convert all tools to LangChain StructuredTools
Then every tool should have a non-empty name and description
And every tool should have a Pydantic args_schema
And every tool should preserve metadata fields from input_schema
72 changes: 72 additions & 0 deletions tests/agentgateway/integration/test_agw_bdd.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,12 +17,15 @@
import asyncio
import os
from typing import Optional
from unittest.mock import AsyncMock

import pytest
from pydantic import BaseModel
from pytest_bdd import scenarios, given, when, then, parsers

from sap_cloud_sdk.agentgateway import AgentGatewayClient, AuthResult, AgentGatewaySDKError
from sap_cloud_sdk.agentgateway._models import MCPTool
from sap_cloud_sdk.agentgateway.converters import mcp_tool_to_langchain

scenarios("agw_auth.feature")

Expand Down Expand Up @@ -52,6 +55,7 @@ def __init__(self):
self.operation_error: Optional[Exception] = None
self.user_token: Optional[str] = None
self.tools: Optional[list[MCPTool]] = None
self.langchain_tools: Optional[list] = None
self.tool_result: Optional[str] = None
self.sample_mcp_tool_name: Optional[str] = None
self.ias_client_id: Optional[str] = None
Expand Down Expand Up @@ -281,3 +285,71 @@ def ias_client_id_non_empty(context: ScenarioContext):
assert context.ias_client_id is not None
assert isinstance(context.ias_client_id, str)
assert context.ias_client_id.strip(), "Expected a non-empty IAS client ID"


# ==================== LANGCHAIN CONVERTER STEPS ====================


@when("I convert all tools to LangChain StructuredTools")
def convert_tools_to_langchain(context: ScenarioContext):
"""Convert every MCPTool from list_mcp_tools to a LangChain StructuredTool."""
assert context.tools is not None, "call list_mcp_tools before converting"
token = context.user_token or ""
context.langchain_tools = [
mcp_tool_to_langchain(t, AsyncMock(return_value="ok"), lambda: token)
for t in context.tools
]


@then("every tool should have a non-empty name and description")
def every_langchain_tool_has_name_and_description(context: ScenarioContext):
"""Verify name and description survive the conversion."""
assert context.langchain_tools is not None
for lc_tool, mcp_tool in zip(context.langchain_tools, context.tools or []):
assert lc_tool.name == mcp_tool.name, (
f"name mismatch: LangChain={lc_tool.name!r}, MCPTool={mcp_tool.name!r}"
)
assert isinstance(lc_tool.description, str) and lc_tool.description.strip(), (
f"Tool '{lc_tool.name}' has empty description"
)


@then("every tool should have a Pydantic args_schema")
def every_langchain_tool_has_args_schema(context: ScenarioContext):
"""Verify every converted tool has a Pydantic BaseModel as args_schema."""
assert context.langchain_tools is not None
for lc_tool in context.langchain_tools:
assert lc_tool.args_schema is not None, (
f"Tool '{lc_tool.name}' has no args_schema"
)
assert isinstance(lc_tool.args_schema, type) and issubclass(
lc_tool.args_schema, BaseModel
), f"Tool '{lc_tool.name}' args_schema is not a Pydantic BaseModel"


@then("every tool should preserve metadata fields from input_schema")
def every_langchain_tool_preserves_metadata(context: ScenarioContext):
"""Verify that description, title, examples, enum, and constraints from the
MCP input_schema appear in the converted Pydantic model_json_schema."""
assert context.langchain_tools is not None
assert context.tools is not None

CHECKED_KEYS = ("description", "title", "examples", "enum", "pattern",
"maxLength", "minimum", "maximum", "format", "default")

for lc_tool, mcp_tool in zip(context.langchain_tools, context.tools):
assert isinstance(lc_tool.args_schema, type) and issubclass(lc_tool.args_schema, BaseModel)
out_props = lc_tool.args_schema.model_json_schema().get("properties", {})
in_props = mcp_tool.input_schema.get("properties", {})

for pname, pdef in in_props.items():
safe = pname.lstrip("_") or pname
assert safe in out_props, (
f"Tool '{mcp_tool.name}': parameter '{pname}' missing from converted schema"
)
out_str = str(out_props[safe])
for key in CHECKED_KEYS:
if key in pdef:
assert f"'{key}'" in out_str or f'"{key}"' in out_str, (
f"Tool '{mcp_tool.name}': '{pname}'.{key} dropped during conversion"
)
Loading
Loading