agent-platform/tests/test_tool_service.py

from pathlib import Path

from tests.conftest import build_fastapi_test_client, prepare_known_service_import


def test_tool_service_post_contract_supports_mcp_connections_and_secrets(
    tmp_path: Path,
    monkeypatch,
) -> None:
    prepare_known_service_import("tool-service")

    from app.bootstrap.app import create_app
    from app.db.models import Base
    from core_db import create_session_factory
    from sqlalchemy import create_engine

    database_url = f"sqlite:///{tmp_path / 'tools.db'}"
    monkeypatch.setenv("AGENT_PLATFORM_DATABASE_URL", database_url)
    engine = create_engine(database_url, connect_args={"check_same_thread": False})
    Base.metadata.create_all(engine)

    app = create_app()
    app.state.session_factory = create_session_factory(engine)
    client = build_fastapi_test_client(app)

    connect_response = client.post(
        "/tools/mcp/connect",
        json={
            "config": {
                "billing_mcp": {
                    "url": "http://127.0.0.1:9090/sse",
                    "headers": {"X-MCP-API-KEY": "secret"},
                    "timeout": 50,
                    "sse_read_timeout": 50,
                    "mcp_tools": [
                        {
                            "name": "query_invoice",
                            "description": "Query invoices",
                            "inputSchema": {"invoiceId": "string"},
                        }
                    ],
                }
            }
        },
    )
    assert connect_response.status_code == 200
    connect_payload = connect_response.json()["data"]
    assert connect_payload["tool"]["name"] == "billing_mcp"
    assert connect_payload["tool"]["toolType"] == "mcp"
    assert connect_payload["connection"]["invokeConfig"]["url"] == "http://127.0.0.1:9090/sse"
    assert connect_payload["connection"]["timeoutMs"] == 50000
    assert connect_payload["discoveredTools"][0]["name"] == "query_invoice"
    assert "code" not in connect_payload["tool"]

    tools_response = client.post(
        "/tools/list",
        json={"page": 1, "pageSize": 20, "keyword": "billing"},
    )
    assert tools_response.status_code == 200
    assert tools_response.json()["data"]["total"] == 1

    update_tool_response = client.post(
        "/tools/update",
        json={
            "toolId": connect_payload["tool"]["id"],
            "name": "Billing MCP Updated",
        },
    )
    assert update_tool_response.status_code == 200
    assert update_tool_response.json()["data"]["name"] == "Billing MCP Updated"

    connections_response = client.post(
        "/tools/connections/list",
        json={"page": 1, "pageSize": 20, "toolId": connect_payload["tool"]["id"]},
    )
    assert connections_response.status_code == 200
    assert (
        connections_response.json()["data"]["items"][0]["id"]
        == connect_payload["connection"]["id"]
    )

    update_connection_response = client.post(
        "/tools/connections/update",
        json={
            "connectionId": connect_payload["connection"]["id"],
            "timeoutMs": 60000,
        },
    )
    assert update_connection_response.status_code == 200
    assert update_connection_response.json()["data"]["timeoutMs"] == 60000

    credential_response = client.post(
        "/tools/credentials/create",
        json={
            "name": "Billing MCP Key",
            "credentialType": "api_key",
            "secretJson": {"apiKey": "secret"},
        },
    )
    assert credential_response.status_code == 200
    credential_payload = credential_response.json()["data"]
    assert credential_payload["credentialType"] == "api_key"

    reveal_response = client.post(
        "/tools/credentials/reveal",
        json={"credentialId": credential_payload["id"]},
    )
    assert reveal_response.status_code == 200
    assert reveal_response.json()["data"]["secretJson"] == {"apiKey": "secret"}

    update_credential_response = client.post(
        "/tools/credentials/update",
        json={
            "credentialId": credential_payload["id"],
            "metadataJson": {"owner": "billing"},
        },
    )
    assert update_credential_response.status_code == 200
    assert update_credential_response.json()["data"]["metadataJson"]["owner"] == "billing"

    binding_response = client.post(
        "/tools/bindings/create",
        json={
            "appId": "agent_support",
            "toolVersionId": connect_payload["connection"]["id"],
            "credentialId": credential_payload["id"],
            "configJson": {"scope": "billing"},
        },
    )
    assert binding_response.status_code == 200
    binding_payload = binding_response.json()["data"]
    assert binding_payload["toolVersionId"] == connect_payload["connection"]["id"]
    assert "enabled" not in binding_payload

    update_binding_response = client.post(
        "/tools/bindings/update",
        json={
            "bindingId": binding_payload["id"],
            "configJson": {"scope": "billing:read"},
        },
    )
    assert update_binding_response.status_code == 200
    assert update_binding_response.json()["data"]["configJson"]["scope"] == "billing:read"

    bindings_response = client.post(
        "/tools/bindings/list",
        json={"page": 1, "pageSize": 20, "appId": "agent_support"},
    )
    assert bindings_response.status_code == 200
    assert bindings_response.json()["data"]["total"] == 1

    delete_binding_response = client.post(
        "/tools/bindings/delete",
        json={"bindingId": binding_payload["id"]},
    )
    assert delete_binding_response.status_code == 200
    assert delete_binding_response.json()["data"]["deleted"] is True

    delete_connection_response = client.post(
        "/tools/connections/delete",
        json={"connectionId": connect_payload["connection"]["id"]},
    )
    assert delete_connection_response.status_code == 200
    assert delete_connection_response.json()["data"]["deleted"] is True

    delete_tool_response = client.post(
        "/tools/delete",
        json={"toolId": connect_payload["tool"]["id"]},
    )
    assert delete_tool_response.status_code == 200
    assert delete_tool_response.json()["data"]["deleted"] is True