Long-running HTTP services

Use @app.inference() when a process should keep running and receive HTTP requests through a CGC URL. The workflow API creates the resource, waits when required, configures its ports, and returns a DeployedEndpoint handle.

This example deploys a small text analysis service from a custom Python image.

Define the service

from cgc.sdk import App, Image, ShellCommand


RESOURCE_NAME = "text-analysis-service"
PORT = 8000
PORT_NAME = "api"

app = App(
    name="http-service-workflow",
    image=Image.from_registry("python:3.11-slim"),
)


@app.inference(
    entity="custom",
    name=RESOURCE_NAME,
    port=PORT,
    port_name=PORT_NAME,
    cpu=1,
    memory=2,
)
def text_analysis_service():
    code = r"""
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json


class Handler(BaseHTTPRequestHandler):
    def _send_json(self, status_code, payload):
        body = json.dumps(payload).encode("utf-8")
        self.send_response(status_code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if self.path in {"/health", "/api/health"}:
            self._send_json(200, {"status": "ok"})
            return
        self._send_json(404, {"error": "not found", "path": self.path})

    def do_POST(self):
        if self.path not in {"/analyze", "/api/analyze"}:
            self._send_json(404, {"error": "not found", "path": self.path})
            return

        length = int(self.headers.get("Content-Length", "0"))
        payload = json.loads(self.rfile.read(length) or b"{}")
        text = payload.get("text", "")
        words = text.split()
        self._send_json(
            200,
            {
                "input": text,
                "characters": len(text),
                "words": len(words),
                "uppercase": text.upper(),
                "first_five_words": words[:5],
            },
        )

    def log_message(self, format, *args):
        print(format % args, flush=True)


ThreadingHTTPServer(("0.0.0.0", 8000), Handler).serve_forever()
"""
    return ShellCommand(executable=("python", "-c", code))

entity="custom" means that CGC runs the image selected in App. The decorated function runs locally and returns the command that CGC starts inside the container. ShellCommand keeps the executable and its arguments structured and quotes them when it renders the final shell command.

The server must listen on 0.0.0.0 and on the same port as port. In this example it listens on 8000.

Install startup dependencies

When the selected image needs additional packages, install them in ShellCommand.prelude before starting the main process:

return ShellCommand(
    executable=("python", "-m", "my_service"),
    arguments=("--host 0.0.0.0", "--port 8000"),
    prelude="python -m pip install --quiet fastapi 'uvicorn[standard]'",
)

The prelude runs on every container start. Package installation is not persisted in the image. For stable and fast startup, put fixed dependencies in your image; use prelude for small runtime additions and development workflows.

Deploy and call the service

endpoint = text_analysis_service.deploy(
    wait=True,
    reuse_existing=False,
    resource_timeout_seconds=600,
)

print("Endpoint:", endpoint.url)
endpoint.wait_http("/health", timeout_seconds=300)

result = endpoint.post_json(
    "/analyze",
    {"text": "CGC SDK workflows can run HTTP services."},
)
print(result)

deploy() first creates the resource and then configures the declared port. wait=True waits for the resource and refreshes its ingress URL. reuse_existing=False raises on a name conflict. True reuses the resource without updating its image, command or environment.

wait_http() waits until the application itself accepts requests. This is different from wait=True: a resource can be Running before its HTTP server is ready.

The endpoint URL may already contain the ingress path prefix derived from port_name. wait_http() and post_json() append the supplied path to that URL. The application must expose routes that match the resulting path. The complete example accepts both unprefixed routes such as /health and prefixed routes such as /api/health.

Custom Image Entrypoint Arguments

Instead of a shell command or ShellCommand, a custom inference function may return a list of arguments:

from cgc.sdk import App, Image

args_app = App("argument-demo", image=Image.from_registry("python:3.12-slim"))

@args_app.inference(name="argument-demo", port=8000, port_name="http")
def serve_with_arguments():
    return ["python", "-m", "http.server", "8000"]

plan = serve_with_arguments.deploy(dry_run=True)

The backend parses the list. See the API reference for space and repeated-key limitations. Use ShellCommand for complex commands.

Add more ports

Use ServicePort when one resource listens on more than one port:

from cgc.sdk import App, Image, ServicePort, ShellCommand

multi_app = App("multi-port-demo", image=Image.from_registry("python:3.12-slim"))

@multi_app.inference(
    name="multi-port-service", port=8000, port_name="http",
    additional_ports=[
        ServicePort("metrics", 9090, ingress=False),
        ServicePort("internal", 5557, ingress=False),
    ],
)
def service():
    code = r"""
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        body = json.dumps({"port": self.server.server_port, "status": "ok"}).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

for port in (8000, 9090, 5557):
    server = ThreadingHTTPServer(("0.0.0.0", port), Handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
threading.Event().wait()
"""
    return ShellCommand(executable=("python", "-u", "-c", code))

plan = service.deploy(dry_run=True)

The SDK creates the resource first, waits for Running, and then adds the primary and additional ports. Port names and numbers must be unique. Each port must be between 1 and 65535. Setting ingress=False keeps a port without a public ingress URL. Declaring a port does not start a listener; the application inside the container must listen on it.

If port configuration fails, the SDK preserves the resource and reports the error so it can be inspected or deleted explicitly.

Additional ports force a wait for Running even with wait=False. With primary ingress=False, do not use endpoint.url, wait_http() or post_json(); internal ports are reachable from workloads in the same namespace.

Use a CGC resource template

For a template, omit image and startup_command. This declaration uses the Nginx template defaults:

from cgc.sdk import App

template_app = App("template-service")
resource = template_app.resource(
    entity="nginx", name="template-nginx", cpu=1, memory=1,
)
plan = resource.create(dry_run=True)

The entity must exist in the selected environment. Pass template arguments through App.resource(arguments=[...]) or return a list from an App.inference(entity=...) function.

Inspect and delete

print(endpoint.status())
print(endpoint.logs(timestamps=True))
endpoint.delete()

Run the complete example with automatic cleanup:

python text_inference.py --delete

See the complete script on the Workflow examples page. For ready-made model APIs, continue with Ready-to-use workflows.