Skip to content

slashed_agent

Class info

Classes

Name Children Inherits
CommandCompleteEvent
llmling_agent.agent.events
Event indicating slash command execution is complete.
    CommandOutputEvent
    llmling_agent.agent.events
    Event for slash command output.
      SlashedAgent
      llmling_agent.agent.slashed_agent
      Wrapper around Agent that handles slash commands in streams.
        StreamingOutputWriter
        llmling_agent.agent.slashed_agent
        OutputWriter that captures command output and emits it as events.

          🛈 DocStrings

          Slash command wrapper for Agent that injects command events into streams.

          SlashedAgent

          Wrapper around Agent that handles slash commands in streams.

          Uses the "commands first" strategy from the ACP adapter: 1. Execute all slash commands first 2. Then process remaining content through wrapped agent 3. If only commands, end without LLM processing

          Source code in src/llmling_agent/agent/slashed_agent.py 
           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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
          class SlashedAgent[TDeps, OutputDataT]:
    """Wrapper around Agent that handles slash commands in streams.

    Uses the "commands first" strategy from the ACP adapter:
    1. Execute all slash commands first
    2. Then process remaining content through wrapped agent
    3. If only commands, end without LLM processing
    """

    def __init__(
        self,
        agent: Agent[TDeps, OutputDataT],
        command_store: CommandStore | None = None,
        *,
        output_writer_factory: Callable[[str], OutputWriter] | None = None,
        context_data_factory: Callable[[], Any] | None = None,
    ) -> None:
        """Initialize with wrapped agent and command store.

        Args:
            agent: The agent to wrap
            command_store: Command store for slash commands(creates default if None)
            output_writer_factory: Optional factory for creating output writers
            context_data_factory: Optional factory for creating command context data
        """
        self.agent = agent
        self.command_store = command_store or _create_default_command_store()
        self._output_writer_factory = output_writer_factory
        self._context_data_factory = context_data_factory

    def _is_slash_command(self, text: str) -> bool:
        """Check if text starts with a slash command.

        Args:
            text: Text to check

        Returns:
            True if text is a slash command
        """
        return bool(SLASH_PATTERN.match(text.strip()))

    async def _execute_slash_command(
        self, command_text: str
    ) -> list[CommandOutputEvent | CommandCompleteEvent]:
        """Execute a single slash command and return events.

        Args:
            command_text: Full command text including slash

        Returns:
            List of events from command execution
        """
        events: list[CommandOutputEvent | CommandCompleteEvent] = []

        parsed = _parse_slash_command(command_text)
        if not parsed:
            logger.warning("Invalid slash command", command=command_text)
            events.append(CommandCompleteEvent(command="unknown", success=False))
            return events

        command_name, args = parsed

        # Create output writer that captures to events
        if self._output_writer_factory:
            output_writer = self._output_writer_factory(command_name)
        else:
            output_writer = StreamingOutputWriter(command_name, events)

        # Create command context
        context_data = (
            self._context_data_factory()
            if self._context_data_factory
            else self.agent.context
        )

        cmd_ctx = self.command_store.create_context(
            data=context_data, output_writer=output_writer
        )

        # Execute command
        command_str = f"{command_name} {args}".strip()
        success = True
        try:
            await self.command_store.execute_command(command_str, cmd_ctx)
        except Exception as e:
            logger.exception("Command execution failed", command=command_name)
            events.append(
                CommandOutputEvent(command=command_name, output=f"Command error: {e}")
            )
            success = False

        # Add completion event
        events.append(CommandCompleteEvent(command=command_name, success=success))
        return events

    async def run_stream(
        self, *prompts: Any, **kwargs: Any
    ) -> AsyncGenerator[SlashedAgentStreamEvent[OutputDataT]]:
        """Run agent with slash command support.

        Separates slash commands from regular prompts, executes commands first,
        then processes remaining content through the wrapped agent.

        Args:
            *prompts: Input prompts (may include slash commands)
            **kwargs: Additional arguments passed to agent.run_stream

        Yields:
            Stream events from command execution and agent processing
        """
        # Separate slash commands from regular content
        commands: list[str] = []
        regular_prompts: list[Any] = []

        for prompt in prompts:
            if isinstance(prompt, str) and self._is_slash_command(prompt):
                logger.debug("Found slash command: %r", prompt)
                commands.append(prompt.strip())
            else:
                regular_prompts.append(prompt)

        # Execute all commands first
        if commands:
            for command in commands:
                logger.info("Processing slash command", command=command)
                command_events = await self._execute_slash_command(command)
                for cmd_event in command_events:
                    yield cmd_event

        # If we have regular content, process it through the agent
        if regular_prompts:
            logger.debug(
                "Processing %d regular prompts through agent", len(regular_prompts)
            )
            async for event in self.agent.run_stream(*regular_prompts, **kwargs):
                yield event

        # If we only had commands and no regular content, we're done
        # (no additional events needed)

    def __getattr__(self, name: str) -> Any:
        """Delegate attribute access to wrapped agent.

        Args:
            name: Attribute name

        Returns:
            Attribute value from wrapped agent
        """
        return getattr(self.agent, name)

          __getattr__ 

          __getattr__(name: str) -> Any

          Delegate attribute access to wrapped agent.

          Parameters:

          Name Type Description Default
          name str

          Attribute name

          		 required

          Returns:

          Type Description
          Any

          Attribute value from wrapped agent
          Source code in src/llmling_agent/agent/slashed_agent.py 
          226
227
228
229
230
231
232
233
234
235
          def __getattr__(self, name: str) -> Any:
    """Delegate attribute access to wrapped agent.

    Args:
        name: Attribute name

    Returns:
        Attribute value from wrapped agent
    """
    return getattr(self.agent, name)

          __init__ 

          __init__(
    agent: Agent[TDeps, OutputDataT],
    command_store: CommandStore | None = None,
    *,
    output_writer_factory: Callable[[str], OutputWriter] | None = None,
    context_data_factory: Callable[[], Any] | None = None,
) -> None

          Initialize with wrapped agent and command store.

          Parameters:

          Name Type Description Default
          agent Agent[TDeps, OutputDataT]

          The agent to wrap

          		 required
          command_store CommandStore | None

          Command store for slash commands(creates default if None)

          		 None
          output_writer_factory Callable[[str], OutputWriter] | None

          Optional factory for creating output writers

          		 None
          context_data_factory Callable[[], Any] | None

          Optional factory for creating command context data

          		 None
          Source code in src/llmling_agent/agent/slashed_agent.py 
           95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
          def __init__(
    self,
    agent: Agent[TDeps, OutputDataT],
    command_store: CommandStore | None = None,
    *,
    output_writer_factory: Callable[[str], OutputWriter] | None = None,
    context_data_factory: Callable[[], Any] | None = None,
) -> None:
    """Initialize with wrapped agent and command store.

    Args:
        agent: The agent to wrap
        command_store: Command store for slash commands(creates default if None)
        output_writer_factory: Optional factory for creating output writers
        context_data_factory: Optional factory for creating command context data
    """
    self.agent = agent
    self.command_store = command_store or _create_default_command_store()
    self._output_writer_factory = output_writer_factory
    self._context_data_factory = context_data_factory

          run_stream async 

          run_stream(
    *prompts: Any, **kwargs: Any
) -> AsyncGenerator[SlashedAgentStreamEvent[OutputDataT]]

          Run agent with slash command support.

          Separates slash commands from regular prompts, executes commands first, then processes remaining content through the wrapped agent.

          Parameters:

          Name Type Description Default
          *prompts Any

          Input prompts (may include slash commands)

          		 ()
          **kwargs Any

          Additional arguments passed to agent.run_stream

          		 {}

          Yields:

          Type Description
          AsyncGenerator[SlashedAgentStreamEvent[OutputDataT]]

          Stream events from command execution and agent processing
          Source code in src/llmling_agent/agent/slashed_agent.py 
          181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
          async def run_stream(
    self, *prompts: Any, **kwargs: Any
) -> AsyncGenerator[SlashedAgentStreamEvent[OutputDataT]]:
    """Run agent with slash command support.

    Separates slash commands from regular prompts, executes commands first,
    then processes remaining content through the wrapped agent.

    Args:
        *prompts: Input prompts (may include slash commands)
        **kwargs: Additional arguments passed to agent.run_stream

    Yields:
        Stream events from command execution and agent processing
    """
    # Separate slash commands from regular content
    commands: list[str] = []
    regular_prompts: list[Any] = []

    for prompt in prompts:
        if isinstance(prompt, str) and self._is_slash_command(prompt):
            logger.debug("Found slash command: %r", prompt)
            commands.append(prompt.strip())
        else:
            regular_prompts.append(prompt)

    # Execute all commands first
    if commands:
        for command in commands:
            logger.info("Processing slash command", command=command)
            command_events = await self._execute_slash_command(command)
            for cmd_event in command_events:
                yield cmd_event

    # If we have regular content, process it through the agent
    if regular_prompts:
        logger.debug(
            "Processing %d regular prompts through agent", len(regular_prompts)
        )
        async for event in self.agent.run_stream(*regular_prompts, **kwargs):
            yield event

          StreamingOutputWriter

          OutputWriter that captures command output and emits it as events.

          Source code in src/llmling_agent/agent/slashed_agent.py 
          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
          class StreamingOutputWriter:
    """OutputWriter that captures command output and emits it as events."""

    def __init__(
        self,
        command: str,
        event_sink: Sequence[CommandOutputEvent | CommandCompleteEvent],
    ) -> None:
        """Initialize with command name and event sink.

        Args:
            command: Name of the command being executed
            event_sink: List to append output events to
        """
        self.command = command
        self.event_sink = list(event_sink)

    async def print(self, message: str = "", **kwargs: Any) -> None:
        """Capture print output as command output event.

        Args:
            message: The message to output
            **kwargs: Additional print arguments (ignored)
        """
        if message:  # Only emit non-empty messages
            event = CommandOutputEvent(command=self.command, output=message)
            self.event_sink.append(event)

          __init__ 

          __init__(
    command: str, event_sink: Sequence[CommandOutputEvent | CommandCompleteEvent]
) -> None

          Initialize with command name and event sink.

          Parameters:

          Name Type Description Default
          command str

          Name of the command being executed

          		 required
          event_sink Sequence[CommandOutputEvent | CommandCompleteEvent]

          List to append output events to

          		 required
          Source code in src/llmling_agent/agent/slashed_agent.py 
          60
61
62
63
64
65
66
67
68
69
70
71
72
          def __init__(
    self,
    command: str,
    event_sink: Sequence[CommandOutputEvent | CommandCompleteEvent],
) -> None:
    """Initialize with command name and event sink.

    Args:
        command: Name of the command being executed
        event_sink: List to append output events to
    """
    self.command = command
    self.event_sink = list(event_sink)

          print async 

          print(message: str = '', **kwargs: Any) -> None

          Capture print output as command output event.

          Parameters:

          Name Type Description Default
          message str

          The message to output

          		 ''
          **kwargs Any

          Additional print arguments (ignored)

          		 {}
          Source code in src/llmling_agent/agent/slashed_agent.py 
          74
75
76
77
78
79
80
81
82
83
          async def print(self, message: str = "", **kwargs: Any) -> None:
    """Capture print output as command output event.

    Args:
        message: The message to output
        **kwargs: Additional print arguments (ignored)
    """
    if message:  # Only emit non-empty messages
        event = CommandOutputEvent(command=self.command, output=message)
        self.event_sink.append(event)