MCPcopy Create free account
hub / github.com/TheSecuredAnalyst/security-suite / stream_ask

Method stream_ask

modules/ai/copilot.py:202–241  ·  view source on GitHub ↗

Stream response to a question. Args: question: Natural language question include_context: Whether to include scan context Yields: Response chunks

(
        self,
        question: str,
        include_context: bool = True,
    )

Source from the content-addressed store, hash-verified

200 async def stream_ask(
201 self,
202 question: str,
203 include_context: bool = True,
204 ) -> AsyncIterator[str]:
205 """Stream response to a question.
206
207 Args:
208 question: Natural language question
209 include_context: Whether to include scan context
210
211 Yields:
212 Response chunks
213 """
214 messages = [Message(role="system", content=SYSTEM_PROMPT)]
215
216 if include_context and self.scan_context:
217 context = self._build_context()
218 messages.append(Message(
219 role="user",
220 content=f"Here is the current scan data:\n\n```json\n{context}\n```"
221 ))
222 messages.append(Message(
223 role="assistant",
224 content="I've analyzed the scan data. What would you like to know?"
225 ))
226
227 messages.extend(self.conversation)
228 messages.append(Message(role="user", content=question))
229
230 client = self._get_client()
231 full_response = ""
232
233 async for chunk in client.stream_chat(messages, temperature=0.3):
234 full_response += chunk
235 yield chunk
236
237 # Update history
238 self.conversation.append(Message(role="user", content=question))
239 self.conversation.append(Message(role="assistant", content=full_response))
240
241 async def analyze(self) -> str:
242 """Get comprehensive analysis of loaded scan results.
243
244 Returns:

Callers

nothing calls this directly

Calls 4

_build_contextMethod · 0.95
_get_clientMethod · 0.95
MessageClass · 0.90
stream_chatMethod · 0.45

Tested by

no test coverage detected