Documentation
¶
Overview ¶
Package calltask executes a tool call apart from the request that asked for it, as an MCP task (#2009).
A client that declares the Tasks extension and calls an export tool (trino_export, api_export, graphql_export) is answered at once with a task handle, and the export runs on the replica that took the call, on a context that outlives the request. The task is a call_tasks row: who started it, where it stands, and once it ends the tool's result or the JSON-RPC error it failed with, so tasks/get is answered by any replica.
While a call runs, its replica stamps the row's heartbeat every few seconds and reads, in the same statement, whether a cancel was asked for: that is how a tasks/cancel another replica took reaches the call. A row whose heartbeat stopped belongs to a replica that is gone, and is read as failed.
The runner sits inside the tools/call chain where mcptask.Middleware hands it a call, so authorization, the gates and the consent prompts have already answered; what runs apart is the rest of the chain and the tool.
Index ¶
- Constants
- type Ending
- type Options
- type PostgresStore
- func (s *PostgresStore) Create(ctx context.Context, t Task) error
- func (s *PostgresStore) FailStale(ctx context.Context, id string, before time.Time, message string, ...) (bool, error)
- func (s *PostgresStore) Finish(ctx context.Context, id string, end Ending) error
- func (s *PostgresStore) Get(ctx context.Context, id string) (*Task, error)
- func (s *PostgresStore) Heartbeat(ctx context.Context, ids []string) ([]string, error)
- func (s *PostgresStore) RequestCancel(ctx context.Context, id string) error
- type Runner
- func (r *Runner) Cancel(ctx context.Context, id string) error
- func (r *Runner) Drain(ctx context.Context) error
- func (r *Runner) Get(ctx context.Context, id string) (*mcptask.Snapshot, error)
- func (*Runner) Prefix() string
- func (r *Runner) Start(ctx context.Context, toolName string, ...) (*mcptask.Snapshot, error)
- func (r *Runner) StartHeartbeat()
- func (r *Runner) Stop(ctx context.Context)
- type Store
- type Task
Constants ¶
const ( // DefaultHeartbeat is how often a replica stamps the calls it runs. DefaultHeartbeat = 2 * time.Second // DefaultPollInterval is how often a client is asked to read a task. DefaultPollInterval = 2 * time.Second // DefaultRetention is how long a task is kept after it is created. DefaultRetention = 7 * 24 * time.Hour )
Timing defaults.
const IDPrefix = "dpt_"
IDPrefix marks a call task's id.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Ending ¶
type Ending struct {
Status string
Message string
Result json.RawMessage
Error json.RawMessage
}
Ending is how a call ended: its terminal status, the message the task carries, and the tool result or the JSON-RPC error, in wire form.
type Options ¶
type Options struct {
Store Store
// Retention is how long a task is kept after it is created; zero is
// DefaultRetention.
Retention time.Duration
// Heartbeat is how often running calls are stamped; zero is
// DefaultHeartbeat.
Heartbeat time.Duration
// PollInterval is the polling interval a client is handed; zero is
// DefaultPollInterval.
PollInterval time.Duration
// Now is the clock; nil is time.Now.
Now func() time.Time
}
Options configure a Runner.
type PostgresStore ¶
type PostgresStore struct {
// contains filtered or unexported fields
}
PostgresStore is the Store over the call_tasks table.
func NewPostgresStore ¶
func NewPostgresStore(db *sql.DB) *PostgresStore
NewPostgresStore returns the Store over db.
func (*PostgresStore) Create ¶
func (s *PostgresStore) Create(ctx context.Context, t Task) error
Create inserts a working task.
func (*PostgresStore) FailStale ¶
func (s *PostgresStore) FailStale(ctx context.Context, id string, before time.Time, message string, rpcErr json.RawMessage) (bool, error)
FailStale fails a working task whose replica stopped reporting.
func (*PostgresStore) Heartbeat ¶
Heartbeat stamps working tasks as alive and returns the ones to cancel.
func (*PostgresStore) RequestCancel ¶
func (s *PostgresStore) RequestCancel(ctx context.Context, id string) error
RequestCancel records a request to stop a working task.
type Runner ¶
type Runner struct {
// contains filtered or unexported fields
}
Runner executes calls as tasks on this replica and answers for the tasks every replica keeps. It is both the mcptask.Detacher and the mcptask.Source for its tasks.
func (*Runner) Cancel ¶
Cancel asks a task's call to stop: at once when this replica runs it, at the running replica's next heartbeat otherwise.
func (*Runner) Drain ¶
Drain waits for the calls this replica runs to finish, or for ctx to end, and takes no new ones. It is the shutdown drain: a call a client follows as a task finishes rather than being cut off.
func (*Runner) Get ¶
Get reports a task to whoever started it or an administrator. A working task whose replica stopped reporting is failed first.
func (*Runner) Start ¶
func (r *Runner) Start(ctx context.Context, toolName string, run func(ctx context.Context) (mcp.Result, error)) (*mcptask.Snapshot, error)
Start creates the call's task and runs the call apart from the request. A call with no identified caller, which nobody could follow, and one that arrives while the replica shuts down, are declined and answered directly.
func (*Runner) StartHeartbeat ¶
func (r *Runner) StartHeartbeat()
StartHeartbeat begins the heartbeat loop. A second call does nothing.
func (*Runner) Stop ¶
Stop cancels the calls still running, each of which records that its server shut down, waits within ctx for them to record it, and stops the heartbeat. A call that does not return when canceled is left to the process's exit; its task then reads as failed once its heartbeat stops.
type Store ¶
type Store interface {
// Create inserts a working task.
Create(ctx context.Context, t Task) error
// Get reads one task, or errTaskNotFound.
Get(ctx context.Context, id string) (*Task, error)
// Finish ends a working task; a task that already ended is left as it
// is.
Finish(ctx context.Context, id string, end Ending) error
// RequestCancel records a request to stop a working task.
RequestCancel(ctx context.Context, id string) error
// Heartbeat stamps the working tasks ids name as alive, and returns the
// ones among them a cancel was requested for.
Heartbeat(ctx context.Context, ids []string) ([]string, error)
// FailStale fails a working task whose heartbeat is older than before,
// reporting whether it did.
FailStale(ctx context.Context, id string, before time.Time, message string, rpcErr json.RawMessage) (bool, error)
}
Store is the call_tasks persistence the runner and the Source use.
type Task ¶
type Task struct {
ID string
ToolName string
UserID string
Status string
StatusMessage string
// Result is the tool's result in its wire form, on a completed task.
Result json.RawMessage
// Error is the JSON-RPC error, on a failed task.
Error json.RawMessage
CreatedAt time.Time
UpdatedAt time.Time
HeartbeatAt time.Time
CancelRequested bool
ExpiresAt time.Time
}
Task is one call_tasks row.