ivcap_ai_tool: A python library for building AI tools for the IVCAP platform

A python library containing various helper and middleware functions to simplify developing AI tools to be deployed on IVCAP.

Note: A template git repositiory using this library can be found on github ivcap-works/ivcap-python-ai-tool-template. You may clone that and start from there.

Register a Tool Function

class Request(BaseModel):
    jschema: str = Field("urn:sd:schema:some_tool.request.1", alias="$schema")
    ...

class Result(BaseModel):
    jschema: str = Field("urn:sd:schema:some_tool.1", alias="$schema")
    ...

def some_tool(req: Request) -> Result:
    """
    Here should go a quite extensive description of what the tool can be
    used for so that an agent can work out if this tool is useful in
    a specific context.

    DO NOT ADD PARAMTER AND RETURN DECRIPTIONS -
       DESCRIBE THEM IN THE `Request` MODEL
    """
    ...

    return Result(...)

add_tool_api_route(app, "/", some_tool, opts=ToolOptions(tags=["Great Tool"]))

Start the Service

app = FastAPI(
  ..
)

if __name__ == "__main__":
    start_tool_server(app, some_tool)

JSON-RPC Middleware

This middleware will convert any POST / with a payload following the JSON-RPC specification to an internal POST /{method} and will return the result formatted according to the JSON-RPC spec.

from ivcap_fastapi import use_json_rpc_middleware

app = FastAPI(
  ..
)

use_json_rpc_middleware(app)

Try-Later Middleware

This middleware is supporting the use case where the execution of a requested service is taking longer than the caller is willing to wait. A typical use case is where the service is itself outsourcing the execution to some other long-running service but may immediately receive a reference to the eventual result.

In this case, raising a TryLaterException will return with a 204 status code and additional information on how to later check back for the result.

from ivcap_fastapi import TryLaterException, use_try_later_middleware
use_try_later_middleware(app)

@app.post("/big_job")
def big_job(req: Request) -> Response:
    jobID, expected_exec_time = scheduling_big_job(req)
    raise TryLaterException(f"/big_job/jobs/{jobID}", expected_exec_time)

@app.get("/big_job/jobs/{jobID}")
def get_job(jobID: str) -> Response:
    resp = find_result_for(job_id)
    return resp