Services

A service is a long-lived object that a workflow starts once and then uses many times: a loaded model, a simulator, a connection pool. AsyncFlow runs it as a service task (function_task(..., service=True)), and Flowgentic's SERVICE_TASK flow type registers one through LangraphIntegration.

Start a service

Decorate the function that starts the service. This example's service is a small simulated server, examples/langgraph-integration/service-task/utils/simple_server.py; it opens no network port:

# SERVICE_TASK that starts a persistent background server
@agents_manager.execution_wrappers.asyncflow(
    flow_type=AsyncFlowType.SERVICE_TASK
)
async def start_api_server(port: int) -> SimpleAsyncServer:
    """
    Starts a background API server.
    Returns the server instance that persists until cancelled.
    """
    print(f"🚀 COLD START: Initializing new server on port {port}...")
    server = SimpleAsyncServer(host="localhost", port=port)
    await server.start()
    print(f"✅ Server {server.instance_id} is now running")
    return server

The decorated function is async: awaiting a call submits the service task and returns its future. Awaiting that future gives the service, and later requests go to the same running instance:

print("\n🔷 Step 1: Start first service (await returns the service handle)")
service_future_1 = await start_api_server(8080)
print(f"   → Got service future handle: {type(service_future_1)}")

# Await the future to get the actual server instance
print("\n🔷 Step 2: Await the future to get server instance")
server_1 = await service_future_1
print(f"   → Server instance: {server_1.instance_id}")
print(f"   → Port: {server_1.port}")

# Make first request
print("\n🔷 Step 3: Make first request")
result_1 = await server_1.handle_request("/api/data", "GET")

Each awaited call starts a new service instance, with its own state:

print("\n🔷 Step 6: Start SECOND service (different port)")
service_future_2 = await start_api_server(8081)
server_2 = await service_future_2

Cancel the future to stop the service task, and release the service's own resources:

print("\n🔷 Step 10: Cancel first service")
service_future_1.cancel()
await server_1.shutdown()
print(f"   ✓ Server 1 ({server_1.instance_id}) shut down")

print("\n🔷 Step 11: Cancel second service")
service_future_2.cancel()
await server_2.shutdown()
print(f"   ✓ Server 2 ({server_2.instance_id}) shut down")

Run the example from the repository root:

uv run --extra examples python -m examples.langgraph-integration.service-task.service-intermittent

It starts two servers, sends four requests to the first and two to the second, and reports one cold start per server:

======================================================================
✅ PROOF: Both services running independently
======================================================================
   Server 1: 4 total requests (counter at 4)
   Server 2: 2 total requests (counter at 2)
   Both maintained separate state and uptime

tests/integration/test_feature_examples.py runs the example and checks these counts.

In the agentic demo

The agentic demo loads its surrogate model as a service on the ai backend, and its tools use that one instance for every request:

# 1. The same resident surrogate service as the deterministic demo --------
@flowgentic(flow_type=AsyncFlowType.SERVICE_TASK, backend="ai")
async def load_surrogate() -> SurrogateService:
    return await SurrogateService.load()

service_future = await load_surrogate()
service = await service_future

The demo's summary records one load and five requests ("loads": 1, "requests": 5). The runner cancels the service's future when the campaign ends.

Service or task

	FUNCTION_TASK	SERVICE_TASK
AsyncFlow call	function_task	function_task(..., service=True)
Awaiting a call returns	The task's result	The task's future; awaiting it gives the service
Each awaited call	Runs the function	Starts a new service instance
Stop it	It ends when the function returns	Cancel the future

Each call runs once. Flowgentic does not restart a service that fails; see Retries and timeouts.

Without Flowgentic, the same service is @flow.function_task(service=True); see LangGraph on AsyncFlow directly.

API

flowgentic.langGraph.execution_wrappers