cf-service-client

Python client library for calling Cogniflow service operations through MCP.

The package exposes a small API for package code and keeps MCP transport details behind a transport abstraction. It does not resolve semantics or import concrete service implementations.

The same package provides the cf presentation adapter:

cf service list
cf service call add_two_numbers --a 2 --b 3 --outputs-json

The CLI uses ServiceClient directly in the same installation and process. Development is the deterministic default; select production explicitly with --runtime production.

Service Boundary Helpers

cf_service_client is the Python-facing helper package for both sides of the universal service boundary.

Consumer side code uses ServiceClient.call_service(...), which builds cf.service.call.v1 envelopes:

from cf_service_client import ServiceClient

client = ServiceClient.for_runtime("development")
result = client.call_service(
    capability="urn:cf:service:Addition",
    inputs={"a": 2, "b": 3},
    parameters={},
)

Runtime selection is resolved by the private cf-runtime-bootstrap executable installed beside the active Python interpreter. The client does not search PATH or fall back to repository or environment configuration. An explicit home can be supplied with ServiceClient.for_runtime("development", home="/absolute/cogniflow/home"). Bootstrap targets must use stdio; HTTP targets are not supported.

call_service normalizes ergonomic dictionaries to the universal cf.service.call.v1 envelope. Service-level results are returned as cf.service.result.v1 content.

Executor side code can use parse_executor_request(...), ok_result(...), and error_result(...) to read cf.service.executor.v1 requests and build cf.service.result.v1 envelopes:

from collections.abc import Mapping

from cf_service_client.executor import parse_executor_request
from cf_service_client.envelopes import error_result, json_output, ok_result


def execute(request: Mapping[str, object]) -> dict[str, object]:
    parsed = parse_executor_request(request)

    if not parsed.operation_iri.endswith("add_two_numbers"):
        return error_result(
            "UNSUPPORTED_OPERATION",
            f"Unsupported operation: {parsed.operation_iri}",
        )

    result = parsed.inputs["a"] + parsed.inputs["b"]
    return ok_result(outputs=[json_output("result", result)])

Validation example:

result = client.call_service(
    capability="urn:cf:service:PackageTemplateConformanceAssessment",
    constraints={
        "https://cogniflow.odea-project.org/cf#hasTemplateName": "cf-python-package-basic",
        "https://cogniflow.odea-project.org/cf#hasTemplateVersion": "1.0.0",
    },
    inputs={"package_path": "stonecastle/cf_concept_service"},
    parameters={"repository_root": "."},
)