Bases: NodeContext[TDeps]
Runtime context for agent execution.
Generically typed with AgentContext[Type of Dependencies]
Source code in src/llmling_agent/agent/context.py
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111 | @dataclass(kw_only=True)
class AgentContext[TDeps = Any](NodeContext[TDeps]):
"""Runtime context for agent execution.
Generically typed with AgentContext[Type of Dependencies]
"""
config: AgentConfig
"""Current agent's specific configuration."""
model_settings: dict[str, Any] = field(default_factory=dict)
"""Model-specific settings."""
data: TDeps | None = None
"""Custom context data."""
tool_name: str | None = None
"""Name of the currently executing tool."""
tool_call_id: str | None = None
"""ID of the current tool call."""
tool_input: dict[str, Any] = field(default_factory=dict)
"""Input arguments for the current tool call."""
@cached_property
def converter(self) -> ConversionManager:
"""Get conversion manager from global config."""
return ConversionManager(self.definition.conversion)
# TODO: perhaps add agent directly to context?
@property
def agent(self) -> Agent[TDeps, Any]:
"""Get the agent instance from the pool."""
assert self.pool, "No agent pool available"
assert self.node_name, "No agent name available"
return self.pool.agents[self.node_name]
async def handle_confirmation(
self,
tool: Tool,
args: dict[str, Any],
) -> ConfirmationResult:
"""Handle tool execution confirmation.
Returns True if:
- No confirmation handler is set
- Handler confirms the execution
"""
provider = self.get_input_provider()
mode = self.config.requires_tool_confirmation
if (mode == "per_tool" and not tool.requires_confirmation) or mode == "never":
return "allow"
history = self.agent.conversation.get_history() if self.pool else []
return await provider.get_tool_confirmation(self, tool, args, history)
async def handle_elicitation(
self,
params: types.ElicitRequestParams,
) -> types.ElicitResult | types.ErrorData:
"""Handle elicitation request for additional information."""
provider = self.get_input_provider()
return await provider.get_elicitation(params)
async def report_progress(self, progress: float, total: float | None, message: str) -> None:
"""Report progress by emitting event into the agent's stream."""
from llmling_agent.agent.events import ToolCallProgressEvent
logger.info("Reporting tool call progress", progress=progress, total=total, message=message)
progress_event = ToolCallProgressEvent(
progress=int(progress),
total=int(total) if total is not None else 100,
message=message,
tool_name=self.tool_name or "",
tool_call_id=self.tool_call_id or "",
tool_input=self.tool_input,
)
await self.agent._event_queue.put(progress_event)
@property
def events(self) -> AgentEventEmitter:
"""Get event emitter with context automatically injected."""
from llmling_agent.agent.event_emitter import AgentEventEmitter
return AgentEventEmitter(self)
|
property
Get the agent instance from the pool.
instance-attribute
Current agent's specific configuration.
cached
property
Get conversion manager from global config.
class-attribute
instance-attribute
data: TDeps | None = None
property
Get event emitter with context automatically injected.
class-attribute
instance-attribute
class-attribute
instance-attribute
tool_call_id: str | None = None
ID of the current tool call.
class-attribute
instance-attribute
Input arguments for the current tool call.
class-attribute
instance-attribute
tool_name: str | None = None
Name of the currently executing tool.
async
handle_confirmation(tool: Tool, args: dict[str, Any]) -> ConfirmationResult
Handle tool execution confirmation.
Returns True if:
- No confirmation handler is set
- Handler confirms the execution
Source code in src/llmling_agent/agent/context.py
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81 | async def handle_confirmation(
self,
tool: Tool,
args: dict[str, Any],
) -> ConfirmationResult:
"""Handle tool execution confirmation.
Returns True if:
- No confirmation handler is set
- Handler confirms the execution
"""
provider = self.get_input_provider()
mode = self.config.requires_tool_confirmation
if (mode == "per_tool" and not tool.requires_confirmation) or mode == "never":
return "allow"
history = self.agent.conversation.get_history() if self.pool else []
return await provider.get_tool_confirmation(self, tool, args, history)
|
async
handle_elicitation(params: ElicitRequestParams) -> ElicitResult | ErrorData
Handle elicitation request for additional information.
Source code in src/llmling_agent/agent/context.py
| async def handle_elicitation(
self,
params: types.ElicitRequestParams,
) -> types.ElicitResult | types.ErrorData:
"""Handle elicitation request for additional information."""
provider = self.get_input_provider()
return await provider.get_elicitation(params)
|
async
report_progress(progress: float, total: float | None, message: str) -> None
Report progress by emitting event into the agent's stream.
Source code in src/llmling_agent/agent/context.py
91
92
93
94
95
96
97
98
99
100
101
102
103
104 | async def report_progress(self, progress: float, total: float | None, message: str) -> None:
"""Report progress by emitting event into the agent's stream."""
from llmling_agent.agent.events import ToolCallProgressEvent
logger.info("Reporting tool call progress", progress=progress, total=total, message=message)
progress_event = ToolCallProgressEvent(
progress=int(progress),
total=int(total) if total is not None else 100,
message=message,
tool_name=self.tool_name or "",
tool_call_id=self.tool_call_id or "",
tool_input=self.tool_input,
)
await self.agent._event_queue.put(progress_event)
|