cognite-typed-functions 0.2.0.post41.dev0

FastAPI-style framework for building type-safe Cognite Functions with automatic OpenAPI schema generation and MCP integration

Cognite Typed Functions

Enterprise-grade framework for building type-safe, composable Cognite Functions with automatic validation, built-in introspection, and AI integration.

Why Cognite Typed Functions?

Standard Cognite Functions require a simple handle(client, data) function, which becomes unwieldy for complex APIs. This framework provides composable architecture, automatic validation, and built-in introspection.

Standard Cognite Function:

def handle(client, data):
    try:
        asset_no = int(data["assetNo"])  # Manual validation
        include_tax = data.get("includeTax", "false").lower() == "true"  # Manual parsing
        # Handle routing manually based on data
        if data.get("action") == "get_item":
            # Implementation here
        elif data.get("action") == "create_item":
            # Different implementation
    except Exception as e:
        return {"error": str(e)}  # Basic error handling

With Typed Functions:

@app.get("/items/{item_id}")
def get_item(client: CogniteClient, item_id: int, include_tax: bool = False) -> ItemResponse:
    """Retrieve an item by ID"""
    # Type validation and coercion handled automatically
    # Clear function signature with proper types
    # Automatic error handling and response formatting

Features

• Type-safe routing - Decorator-based syntax (@app.get(), @app.post(), etc.) with automatic validation
• Async/await support - Write both sync and async handlers for concurrent operations
• Automatic type validation - Recursive conversion of nested data structures with Pydantic models
• OpenAPI schema generation - Auto-generated API documentation
• Built-in introspection - /__schema__, /__routes__, /__health__, /__ping__ endpoints
• Model Context Protocol (MCP) - Native AI tool exposure for LLM integration
• Comprehensive error handling - Structured error responses with detailed information
• Enterprise logging - Isolated logger with dependency injection across all cloud providers
• Composable architecture - Build modular services from reusable middleware apps
• Path and query parameters - Support for dynamic URL parameters and query strings
• Modern Python - Python 3.10+ with union types (x | None), builtin generics
• Full Cognite Functions compatibility - Works with scheduling, secrets, and all deployment methods

Quick Start

Installation

Requirements:

• Python 3.10 or higher
• uv (recommended) or pip

# Install the package (when published)
# pip install cognite-typed-functions

Basic Example

# No typing imports needed - using builtin generic types
from cognite.client import CogniteClient
from pydantic import BaseModel

from cognite_typed_functions import FunctionApp, create_function_service

# Create your app
app = FunctionApp(title="My API", version="1.0.0")

# Define your models
class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

class ItemResponse(BaseModel):
    id: int
    item: Item
    total_price: float

# Define your endpoints
@app.get("/items/{item_id}")
def get_item(
    client: CogniteClient,
    item_id: int,
    include_tax: bool = False
) -> ItemResponse:
    """Retrieve an item by ID"""
    item = Item(
        name=f"Item {item_id}",
        price=100.0,
        tax=10.0 if include_tax else None
    )
    total = item.price + (item.tax or 0)
    return ItemResponse(id=item_id, item=item, total_price=total)

@app.post("/items/")
def create_item(client: CogniteClient, item: Item) -> ItemResponse:
    """Create a new item"""
    new_id = 12345  # Your creation logic here
    total = item.price + (item.tax or 0)
    return ItemResponse(id=new_id, item=item, total_price=total)

@app.post("/process/batch")
def process_batch(client: CogniteClient, items: list[Item]) -> dict:
    """Process multiple items in batch"""
    total_value = sum(item.price + (item.tax or 0) for item in items)
    return {"processed_count": len(items), "total_value": total_value}

# Export the handler for Cognite Functions
handle = create_function_service(app)

Local Development

Test your functions locally using uvicorn before deploying to Cognite:

# dev.py
from cognite_typed_functions.devserver import create_asgi_app
from handler import handle

app = create_asgi_app(handle)

Then run:

uv run uvicorn dev:app --reload

For complete setup instructions, environment variables, and troubleshooting, see Local Development Server Guide.

Error Handling

The framework provides structured error handling with detailed information for debugging:

• RouteNotFound - No matching route found
• ValidationError - Input validation failed
• TypeConversionError - Parameter type conversion failed
• ExecutionError - Function execution failed

All error responses follow a consistent structure:

{
    "success": false,
    "error_type": "ValidationError",
    "message": "Input validation failed: 1 error(s)",
    "details": {"errors": [...]}
}

Success responses are similarly structured:

{
    "success": true,
    "data": {...}  # Your actual response data
}

Logging

The framework provides an enterprise-grade logging solution that works across all cloud providers through dependency injection.

Why Use the Framework Logger?

According to the Cognite Functions documentation, the standard Python logging module is not recommended because it can interfere with the cloud provider's logging infrastructure. This framework provides an isolated logger that:

• Uses Python's standard logging module with familiar API
• Writes directly to stdout (captured by all cloud providers)
• Is completely isolated from other loggers
• Supports standard log levels (DEBUG, INFO, WARNING, ERROR, CRITICAL)
• Can be dependency-injected like client and secrets
• Works with both sync and async handlers

Logger Usage

Add logger: logging.Logger to your function signature:

import logging
from cognite.client import CogniteClient
from cognite_typed_functions import FunctionApp

app = FunctionApp(title="My API", version="1.0.0")

@app.get("/items/{item_id}")
def get_item(client: CogniteClient, logger: logging.Logger, item_id: int) -> dict:
    """Retrieve an item with logging"""
    logger.info(f"Fetching item {item_id}")
    item = fetch_item(item_id)
    logger.debug(f"Item details: {item}")
    return {"id": item_id, "name": item.name}

Log Levels

@app.post("/process/data")
def process_data(client: CogniteClient, logger: logging.Logger, data: dict) -> dict:
    logger.debug("Detailed debug information")      # DEBUG: Detailed diagnostic info
    logger.info("Processing started")               # INFO: General informational messages
    logger.warning("Unexpected value encountered")  # WARNING: Warning messages
    logger.error("Processing failed")               # ERROR: Error messages
    logger.critical("System failure")               # CRITICAL: Critical errors
    return {"status": "processed"}

By default, the logger is configured at INFO level.

Async Handlers

The logger works seamlessly with async handlers:

@app.post("/process/batch")
async def process_batch(
    client: CogniteClient,
    logger: logging.Logger,
    items: list[Item]
) -> dict:
    logger.info(f"Starting batch processing of {len(items)} items")

    async def process_item(item: Item) -> dict:
        logger.debug(f"Processing item: {item.name}")
        result = await process_async(item)
        return result

    results = await asyncio.gather(*[process_item(item) for item in items])
    logger.info(f"Batch processing complete. Processed {len(results)} items")
    return {"processed_count": len(results), "results": results}

Best Practices

• Use appropriate log levels: DEBUG for diagnostics, INFO for normal operation, WARNING for unexpected situations, ERROR for failures, CRITICAL for severe errors
• Don't log sensitive information: Avoid logging credentials, tokens, or personal data
• Use structured logging: Include relevant context in log messages
• Log at key points: Entry points, success paths, and error conditions

Dependency Injection

The framework uses dependency injection to provide framework dependencies (client, secrets, logger, function_call_info) to your handlers. You can also register your own custom dependencies for services like database connections, tracing, caching, or any other resources your application needs.

Dependency Matching Semantics

The framework uses AND semantics for dependency matching, providing clear and predictable behavior:

Framework Dependencies (Strict Matching):

• client - Requires both param_name="client" AND target_type=CogniteClient
• secrets - Requires both param_name="secrets" AND target_type=Mapping (accepts dict, Mapping, dict[str, str], Mapping[str, str], etc.)
• logger - Requires both param_name="logger" AND target_type=logging.Logger
• function_call_info - Requires both param_name="function_call_info" AND target_type=FunctionCallInfo

Custom Dependencies (Flexible Matching):

Note: target_type is always required - this is a typed functions framework.

• Register with target_type only for flexible parameter naming (works with any parameter name)
• Register with both param_name and target_type to require BOTH (strict matching with AND logic)

Registering Custom Dependencies

Register custom dependencies on your app's registry before defining routes:

from cognite_typed_functions import FunctionApp
from opentelemetry import trace
import redis

app = FunctionApp(title="My API", version="1.0.0")

# Type-based registration (flexible parameter names)
app.registry.register(
    provider=lambda ctx: trace.get_tracer("my-cognite-function"),
    target_type=trace.Tracer,
    description="OpenTelemetry tracer for distributed tracing"
)

app.registry.register(
    provider=lambda ctx: redis.Redis.from_url(ctx.get("secrets", {}).get("REDIS_URL")),
    target_type=redis.Redis,
    description="Redis cache connection"
)

# Now use them in your handlers with ANY parameter name
@app.get("/items/{item_id}")
def get_item(
    client: CogniteClient,        # Framework: requires name + type
    tracer: trace.Tracer,          # Custom: matched by type (any name works)
    cache: redis.Redis,            # Custom: matched by type (any name works)
    item_id: int
) -> dict:
    with tracer.start_as_current_span("get_item"):
        # Try cache first
        cached = cache.get(f"item:{item_id}")
        if cached:
            return json.loads(cached)

        # Fetch from CDF and cache
        item = client.assets.retrieve(id=item_id)
        cache.set(f"item:{item_id}", json.dumps(item.dump()), ex=3600)
        return item.dump()

# Different parameter names work too!
@app.get("/users/{user_id}")
def get_user(
    client: CogniteClient,
    my_tracer: trace.Tracer,       # Different name, same type - works!
    redis_conn: redis.Redis,       # Different name, same type - works!
    user_id: int
) -> dict:
    # Same dependencies, different parameter names
    return {"user_id": user_id}

Context-Aware Dependencies

Provider functions receive a context dictionary with client, secrets, and function_call_info:

class MyAPIClient:
    def __init__(self, api_key: str, environment: str):
        self.api_key = api_key
        self.environment = environment

    def fetch_data(self):
        # Implementation here
        pass

# Register with type-based matching
app.registry.register(
    provider=lambda ctx: MyAPIClient(
        api_key=ctx.get("secrets", {}).get("API_KEY"),
        environment=ctx.get("secrets", {}).get("ENV", "production")
    ),
    target_type=MyAPIClient,
    description="External API client with credentials"
)

@app.post("/sync-data")
def sync_data(
    client: CogniteClient,      # Framework dependency
    api_client: MyAPIClient,    # Custom dependency (any parameter name works)
    data: dict
) -> dict:
    # api_client is initialized with secrets from context
    external_data = api_client.fetch_data()
    # Process and return
    return {"synced": True}

Registry Sharing in Composed Apps

When you compose multiple apps, custom dependencies from your main app are automatically shared with MCP and introspection apps:

from cognite_typed_functions import create_function_service
from cognite_typed_functions.mcp import create_mcp_app
from cognite_typed_functions.introspection import create_introspection_app

# Main app with custom dependencies
app = FunctionApp(title="My API", version="1.0.0")
app.registry.register(
    provider=lambda ctx: trace.get_tracer("my-app"),
    target_type=trace.Tracer,
    description="OpenTelemetry tracer"
)

# Create extension apps
mcp_app = create_mcp_app()
introspection_app = create_introspection_app()

# Compose apps - main app's registry (with tracer) is shared!
handle = create_function_service(introspection_app, mcp_app, app)

# Now MCP tools can also use 'tracer' if they declare it with the type annotation
@mcp_app.tool("Get item with tracing")
def get_item_tool(client: CogniteClient, tracer: trace.Tracer, item_id: int) -> dict:
    with tracer.start_as_current_span("mcp_get_item"):
        # Tracer from main app is available here!
        return client.assets.retrieve(id=item_id).dump()

Built-in Dependencies

The framework provides these dependencies by default:

• client: CogniteClient - Requires both parameter name client AND type annotation CogniteClient
• secrets: Mapping[str, str] - Requires both parameter name secrets AND Mapping-compatible type (accepts dict, Mapping, dict[str, str], Mapping[str, str], etc.)
• logger: logging.Logger - Requires both parameter name logger AND type annotation logging.Logger
• function_call_info: FunctionCallInfo - Requires both parameter name function_call_info AND type annotation FunctionCallInfo

These are always available and can be combined with your custom dependencies.

Important: Framework dependencies enforce strict naming to match Cognite Functions conventions. You must use the exact parameter names with proper type annotations.

Async Support

The framework supports both synchronous and asynchronous route handlers, enabling efficient concurrent code when needed.

Why Use Async?

Async handlers are particularly useful for:

• Concurrent API calls - Fetch data from multiple sources simultaneously
• I/O-bound operations - Database queries, file operations, network requests
• Parallel processing - Process multiple items concurrently
• External service integration - Call multiple external APIs in parallel

Basic Async Usage

Simply declare your route handler as async def instead of def:

import asyncio
from cognite_typed_functions import FunctionApp

app = FunctionApp(title="Async API", version="1.0.0")

# Synchronous handler (traditional)
@app.get("/items/{item_id}")
def get_item(client: CogniteClient, item_id: int) -> ItemResponse:
    """Synchronous data retrieval"""
    # Your sync logic here
    return ItemResponse(...)

# Asynchronous handler (new!)
@app.get("/items/{item_id}/async")
async def get_item_async(client: CogniteClient, item_id: int) -> ItemResponse:
    """Asynchronous data retrieval with concurrent operations"""
    # Use await for async operations
    result = await fetch_data_async(item_id)
    return ItemResponse(...)

Concurrent Operations Example

The real power of async comes from running multiple operations concurrently:

@app.get("/items/{item_id}/details")
async def get_item_with_details(client: CogniteClient, item_id: int) -> dict:
    """Fetch item data from multiple sources concurrently"""

    # Define async operations
    async def fetch_item_info():
        # Simulate API call
        await asyncio.sleep(0.1)
        return {"name": f"Item {item_id}", "price": 100.0}

    async def fetch_inventory():
        # Simulate another API call
        await asyncio.sleep(0.1)
        return {"stock": 50, "warehouse": "A"}

    async def fetch_reviews():
        # Simulate yet another API call
        await asyncio.sleep(0.1)
        return {"rating": 4.5, "count": 120}

    # Execute all operations concurrently (not sequentially!)
    item_info, inventory, reviews = await asyncio.gather(
        fetch_item_info(),
        fetch_inventory(),
        fetch_reviews()
    )

    return {
        "item": item_info,
        "inventory": inventory,
        "reviews": reviews
    }

Batch Processing with Async

Process multiple items concurrently for better performance:

@app.post("/process/batch/async")
async def process_batch_async(client: CogniteClient, items: list[Item]) -> dict:
    """Process multiple items concurrently"""

    async def process_item(item: Item) -> dict:
        """Process a single item asynchronously"""
        # Simulate async processing (e.g., API call, database query)
        await asyncio.sleep(0.01)
        total = item.price + (item.tax or 0)
        return {"name": item.name, "total": total}

    # Process all items concurrently
    results = await asyncio.gather(*[process_item(item) for item in items])

    total_value = sum(result["total"] for result in results)
    return {
        "processed_count": len(items),
        "total_value": total_value,
        "items": results
    }

How It Works

The framework automatically detects whether your handler is sync or async:

• Async handlers (async def) are awaited directly for native async execution
• Sync handlers (def) are run on a thread pool to avoid blocking the event loop
• MCP tools support both sync and async handlers seamlessly
• App composition works with any mix of sync and async handlers

Performance Considerations

When async helps:

• Multiple I/O operations that can run in parallel
• External API calls that can be concurrent
• Database queries that can be batched

When sync is fine:

• Simple CPU-bound calculations
• Single database/API call
• Straightforward data transformations

Note: Since Cognite Functions don't handle concurrent requests within the same process (each function call gets its own compute instance), async is primarily beneficial for concurrent operations within a single request, not for handling multiple requests simultaneously.

Mixing Sync and Async

You can freely mix sync and async handlers in the same app:

app = FunctionApp(title="Mixed API", version="1.0.0")

@app.get("/simple")
def simple_endpoint(client: CogniteClient) -> dict:
    """Simple sync endpoint"""
    return {"status": "ok"}

@app.get("/complex")
async def complex_endpoint(client: CogniteClient) -> dict:
    """Complex async endpoint with concurrent operations"""
    results = await asyncio.gather(
        fetch_data_1(),
        fetch_data_2(),
        fetch_data_3()
    )
    return {"results": results}

# Both work seamlessly in the same app!
handle = create_function_service(app)

Type Safety and Validation

The framework provides comprehensive type safety with automatic validation and conversion:

• Input validation - Pydantic models validate request data
• Output validation - Response models ensure consistent output format
• Type coercion - Automatic conversion of string parameters to correct types
• Detailed error messages - Validation errors include precise paths for debugging

Basic Type Conversions

• str → int / float / bool (accepts "true", "1", "yes", "on")
• dict → BaseModel - Automatic instantiation with validation
• list[dict] → list[BaseModel] - Converts lists of dictionaries to model instances

Recursive Type Conversions

The framework handles arbitrarily nested combinations:

# Complex nested types supported
dict[str, BaseModel]                    # Dict with model values
Optional[BaseModel]                     # Optional models
Union[BaseModel, str]                   # Union types with fallback
list[dict[str, BaseModel]]              # List of dicts of models
dict[str, list[BaseModel]]              # Dict containing lists of models

# Real-world example
class User(BaseModel):
    name: str
    age: int

class Team(BaseModel):
    name: str
    leader: User                        # Nested model
    members: list[User]                 # List of models

@app.post("/teams")
def create_team(client: CogniteClient, team: Team) -> TeamResponse:
    # Input automatically converted:
    # {
    #   "name": "Engineering",
    #   "leader": {"name": "Alice", "age": 30},      # → User instance
    #   "members": [                                 # → list[User]
    #     {"name": "Bob", "age": 25},                # → User instance
    #     {"name": "Carol", "age": 28}               # → User instance
    #   ]
    # }
    return TeamResponse(id=team.name, members_count=len(team.members))

Type Annotation Compatibility

The framework supports both legacy and modern Python type annotation syntaxes:

# Union Types - both work identically
from typing import Union
def process(client: CogniteClient, data: Union[User, str]) -> Response: ...
def process(client: CogniteClient, data: User | str) -> Response: ...

# Optional Types - both work identically
from typing import Optional
def get_user(client: CogniteClient, user: Optional[User]) -> Response: ...
def get_user(client: CogniteClient, user: User | None) -> Response: ...

# Collection Types - both work identically
from typing import List, Dict
def process_items(client: CogniteClient, items: List[Item]) -> Dict[str, int]: ...
def process_items(client: CogniteClient, items: list[Item]) -> dict[str, int]: ...

Introspection

One of the key challenges with standard Cognite Functions is that they become "black boxes" after deployment. This framework solves that problem with built-in introspection endpoints.

Available Endpoints

• /__schema__ - Returns the complete OpenAPI 3.0 schema for all composed apps
• /__routes__ - Returns a summary of all available routes with descriptions
• /__health__ - Returns health status and comprehensive app information
• /__ping__ - Simple connectivity check for monitoring and pre-warming

Benefits

• No more redeployments just to check function signatures
• AI tools can discover and generate code for your functions
• Team members can easily understand deployed functions
• Documentation stays in sync with implementation

Example Usage

# Get complete API documentation
curl "https://your-function-url" -d '{"path": "/__schema__", "method": "GET"}'

# List all available endpoints
curl "https://your-function-url" -d '{"path": "/__routes__", "method": "GET"}'

# Check function health
curl "https://your-function-url" -d '{"path": "/__health__", "method": "GET"}'

Cross-App Introspection

When composing multiple apps, introspection endpoints show routes from all apps:

// /__schema__ response includes routes from all apps
{
  "info": {
    "title": "Your Main App",      // From main_app (last in composition)
    "version": "1.0.0"
  },
  "paths": {
    "/__mcp_tools__": {...},       // From MCP app
    "/__schema__": {...},          // From introspection app
    "/your/business/route": {...}  // From main business app
  }
}

Model Context Protocol (MCP)

The framework includes built-in Model Context Protocol support, enabling AI assistants to discover and use your Cognite Functions as tools.

MCP Endpoints

• /__mcp_tools__ - List all available MCP tools with their schemas
• /__mcp_call__/{tool_name} - Execute a specific MCP tool by name

Usage

from cognite_typed_functions import FunctionApp, create_function_service
from cognite_typed_functions.mcp import create_mcp_server
from cognite_typed_functions.introspection import create_introspection_app

# Create your main business app
app = FunctionApp(title="Asset Management API", version="1.0.0")

@app.get("/items/{item_id}")
def get_item(client: CogniteClient, item_id: int) -> ItemResponse:
    """Retrieve an item by ID"""
    # Your implementation here

# Create MCP server from your app
mcp = create_mcp_server(app, "asset-management-tools")

# Use @mcp.tool() decorator to expose specific routes to AI
@mcp.tool()
@app.get("/items/{item_id}")
def get_item_for_ai(client: CogniteClient, item_id: int) -> ItemResponse:
    """AI-accessible version of get_item"""
    return get_item(client, item_id)

# Create introspection app
introspection = create_introspection_app()

# Compose all apps
handle = create_function_service(mcp, introspection, app)

MCP Capabilities

• Selective exposure - Use @mcp.tool() to choose which endpoints are accessible to AI
• Automatic schema generation - AI gets JSON schemas for all parameters and responses
• Built-in validation - Input validation happens automatically
• Tool discovery - AI can discover available tools via /__mcp_tools__
• Tool execution - AI can call tools via /__mcp_call__/{tool_name}

App Composition

Note: This is an advanced feature for framework extensibility. Most developers won't need to use app composition directly - it's primarily used internally for features like MCP integration and introspection endpoints. For typical use cases, simply create one FunctionApp and add your routes.

The framework supports composing multiple apps together to create modular services.

Composition Architecture

Apps are composed using left-to-right evaluation for routing, with the last app providing metadata (title, version):

from cognite_typed_functions import FunctionApp, create_function_service
from cognite_typed_functions.introspection import create_introspection_app

# Create individual apps
introspection_app = create_introspection_app()
main_app = FunctionApp("Asset Management API", "2.1.0")

@main_app.get("/assets/{asset_id}")
def get_asset(client: CogniteClient, asset_id: int) -> dict:
    return {"id": asset_id, "name": f"Asset {asset_id}"}

# Compose apps
handle = create_function_service(introspection_app, main_app)

Composition Benefits

• Cross-app introspection - /__schema__ and /__routes__ show routes from all composed apps
• Unified metadata - Uses the last app (main business app) for title/version
• Routing precedence - Earlier apps in composition handle routes first
• Modular design - Separate system utilities from business logic

API Reference

FunctionApp

The main application class for building composable Cognite Function services.

app = FunctionApp(title="My API", version="1.0.0")

Decorators

• @app.get(path) - Handle GET requests (data retrieval)
• @app.post(path) - Handle POST requests (create resources). Also to handle generic operations that don't fit REST semantics (batch processing, calculations, transformations)
• @app.put(path) - Handle PUT requests (update/replace resources)
• @app.delete(path) - Handle DELETE requests (remove resources)

When to Use Each Decorator

Use @app.get() for:

• Retrieving data: @app.get("/assets/{asset_id}")
• Listing resources: @app.get("/assets")
• Health checks: @app.get("/health")

Use @app.post() for:

• Creating new resources: @app.post("/assets")
• Uploading data: @app.post("/files/upload")
• Batch processing: @app.post("/process/batch")
• Complex calculations: @app.post("/calculate/metrics")
• Data transformations: @app.post("/transform/timeseries")
• Operations that don't fit CRUD patterns

Use @app.put() for:

• Updating existing resources: @app.put("/assets/{asset_id}")
• Replacing configurations: @app.put("/settings")

Use @app.delete() for:

• Removing resources: @app.delete("/assets/{asset_id}")
• Cleanup operations: @app.delete("/cache")

Parameters

All endpoint functions must accept client: CogniteClient as the first parameter. Additional parameters can be:

• Path parameters: {item_id} in the URL path
• Query parameters: URL query string parameters
• Request body: Pydantic models for POST/PUT requests

Parameter Injection and Override Behavior:

The framework automatically injects certain parameters into endpoint functions. Currently, this includes the client: CogniteClient parameter which is automatically provided as the first parameter to all endpoint functions.

If users provide parameters with the same names in their request arguments (through query parameters, path parameters, or request body), the framework will attempt to override the injected values with strict type validation. The provided values must be convertible to the expected parameter types, or a validation error will be raised.

@app.get("/example/{count}")
def example_endpoint(client: CogniteClient, name: str, count: int) -> dict:
    # client is normally the injected CogniteClient instance

    # Parameter override examples:
    # {"name": "test", "count": "123"}         ✅ Works - valid conversions
    # {"name": "test", "count": "abc"}         ❌ ValidationError - cannot convert "abc" to int
    # {"client": "invalid", "name": "test"}    ❌ ValidationError - cannot convert string to CogniteClient

    return {"message": f"Hello {name}, count: {count}"}

create_function_service

Creates a service handler from one or more composed apps. This is the main entry point for converting your FunctionApp instances into a service compatible with Cognite Functions.

from cognite_typed_functions import create_function_service

# Single app
handle = create_function_service(app)

# Multiple apps (composition)
handle = create_function_service(mcp_app, introspection_app, main_app)

Function Signature

def create_function_service(*apps: FunctionApp) -> Handler:
    """Create handler for single app or composed apps.

    Args:
        apps: Single FunctionApp or sequence of FunctionApps to compose.
              For composed apps, routing tries each app left-to-right until one matches.

    Returns:
        Handler function compatible with Cognite Functions
    """

App Composition Rules

1. Routing Order: Apps are tried left-to-right for route matching
2. Metadata Source: The last app provides title/version for schemas and health checks
3. Context Sharing: All apps get composition context, can override method if they need it
4. Simplicity: Clean method calls, no complex protocols or type checking overhead

Composition Patterns

System + Business Pattern:

# Introspection provides system endpoints, main_app provides business logic
handle = create_function_service(introspection_app, main_app)

Full Stack Pattern:

# Complete composition: AI tools + debugging + business logic
handle = create_function_service(mcp_app, introspection_app, main_app)

Development Pattern:

# Add debugging capabilities to any existing handler
debug_handle = create_function_service(introspection_app, existing_app)

Request Format

Cognite Functions receive requests in this format:

{
    "path": "/items/123?include_tax=true&q=search",
    "method": "GET",
    "body": {...}  # Optional request body
}

Response Format

All responses follow a structured format:

# Success response
{
    "success": true,
    "data": {...}  # Your actual response data
}

# Error response
{
    "success": false,
    "error_type": "ValidationError",
    "message": "Input validation failed: 1 error(s)",
    "details": {"errors": [...]}
}

Architecture

The framework is organized into several modules:

• app.py - Core application class and request handling with FastAPI-style decorators
• models.py - Shared Pydantic models for responses, errors, and request parsing
• schema.py - OpenAPI schema generation utilities
• introspection.py - Core introspection endpoints (schema, routes, health)
• mcp.py - Model Context Protocol integration and AI tool exposure
• devserver/ - Local development server with ASGI adapter for uvicorn

Key Components

1. FunctionApp - Main application class with FastAPI-style decorators
2. *create_function_service(apps) - Creates composed handler from multiple apps
3. App Composition System:
   • Method override: Apps override set_context() to access routes, other apps, and shared registry
   • Automatic context provision: Context provided to all apps during composition
   • Registry sharing: Main app's dependency registry is shared with all composed apps
   • Left-to-right routing: Earlier apps in composition handle routes first
   • Last-app metadata: Uses final app for title/version in schemas
4. SchemaGenerator - Generates unified OpenAPI documentation across all apps
5. Built-in Apps:
   • IntrospectionApp: Provides /__schema__, /__routes__, /__health__ endpoints
   • MCPApp: Provides /__mcp_tools__, /__mcp_call__/* endpoints
6. Request Processing Pipeline:
   • Parse request data and URL
   • Try each composed app in order (left-to-right evaluation)
   • Find matching route in current app
   • Validate and coerce parameters with recursive type conversion
   • Execute function with automatic error handling
   • Format response with structured success/error format

Examples

The framework includes a complete example in examples/handler.py demonstrating:

• Type-safe routing with decorator syntax
• MCP integration for AI tool exposure
• Built-in introspection endpoints
• Async handler support
• Composable app architecture

Limitations

• The framework does not support multiple body parameters. This may be supported in future versions.

Development & Contributing

Project Structure

cognite-typed-functions/
├── src/
│   └── cognite_typed_functions/
│       ├── app.py              # Core FunctionApp class and decorators
│       ├── convert.py          # Type conversion and argument processing
│       ├── formatting.py       # Formatting utilities
│       ├── models.py           # Pydantic models and type definitions
│       ├── routing.py          # Route matching and management
│       ├── schema.py           # OpenAPI schema generation
│       ├── introspection.py    # Built-in introspection endpoints
│       ├── logger.py           # Enterprise logging utilities
│       ├── mcp.py              # Model Context Protocol integration
│       └── devserver/          # Local development server
│           ├── __init__.py     # Module exports
│           ├── asgi.py         # ASGI adapter for uvicorn
│           └── auth.py         # CogniteClient authentication
├── docs/
│   └── dev-server.md           # Local development guide
├── examples/
│   ├── handler.py              # Complete example with MCP integration
│   └── dev.py                  # Local dev server example
├── tests/                      # Comprehensive test suite
└── pyproject.toml              # Project configuration and dependencies

Running Tests

# Run all tests
uv run pytest

# Run with verbose output
uv run pytest -v

# Run with coverage
uv run pytest --cov=cognite_typed_functions

Contributing

1. Fork the repository
2. Create a feature branch
3. Make your changes with appropriate tests
4. Ensure all tests pass
5. Submit a pull request

License

This project is licensed under the Apache License 2.0 - see the LICENSE file for details.

Acknowledgments

• Built specifically for Cognite Data Fusion Functions platform
• Decorator routing syntax inspired by FastAPI
• Data validation powered by Pydantic