acp_sdk.server.utils module

async acp_sdk.server.utils.async_request_with_retry(request_func, max_retries=5, backoff_factor=1)[source]
Parameters:
  • request_func (Callable[[AsyncClient], Coroutine[Any, Any, Response]])

  • max_retries (int)

  • backoff_factor (float)

Return type:

dict[str, Any]

acp_sdk.server.utils.encode_sse(model)[source]
Parameters:

model (BaseModel)

Return type:

str

async acp_sdk.server.utils.stream_sse(bundle)[source]
Parameters:

bundle (RunBundle)

Return type:

AsyncGenerator[str]