Interactive REPL session.
| 234 | |
| 235 | |
| 236 | class OpenOSINTRepl: |
| 237 | """Interactive REPL session.""" |
| 238 | |
| 239 | def __init__( |
| 240 | self, |
| 241 | api_key: str | None = None, |
| 242 | provider: str = "anthropic", |
| 243 | ollama_model: str = "llama3.2", |
| 244 | ollama_host: str = "http://localhost:11434", |
| 245 | is_pdf_disabled: bool = False, |
| 246 | ) -> None: |
| 247 | self._api_key = api_key or os.environ.get("ANTHROPIC_API_KEY", "") |
| 248 | self._provider = provider |
| 249 | self._ollama_model = ollama_model |
| 250 | self._ollama_host = ollama_host |
| 251 | self._is_pdf_disabled = is_pdf_disabled |
| 252 | |
| 253 | if provider == "ollama": |
| 254 | self._agent: OpenOSINTAgent | OllamaAgent = OllamaAgent( |
| 255 | model=ollama_model, |
| 256 | host=ollama_host, |
| 257 | ) |
| 258 | self._display_model = ollama_model |
| 259 | else: |
| 260 | self._agent = OpenOSINTAgent(api_key=self._api_key) |
| 261 | self._display_model = "claude-sonnet-4-20250514" |
| 262 | |
| 263 | self._last_response: str = "" |
| 264 | self._session_start: datetime = datetime.now() |
| 265 | self._session_prompts: list[str] = [] |
| 266 | self._session_tools: list[str] = [] |
| 267 | self._session_targets: list[str] = [] |
| 268 | self._session_report_path: str = "" |
| 269 | self._session: PromptSession = PromptSession( |
| 270 | history=FileHistory(str(Path.home() / ".openosint_history")), |
| 271 | style=PROMPT_STYLE, |
| 272 | ) |
| 273 | |
| 274 | def _get_prompt_tokens(self) -> HTML: |
| 275 | return HTML("<prompt>openosint</prompt> <prompt-text>❯</prompt-text> ") |
| 276 | |
| 277 | async def _handle_tool_call(self, name: str, args: dict[str, Any]) -> None: |
| 278 | _print_tool_call(name, args) |
| 279 | |
| 280 | async def _run_investigation(self, user_input: str) -> None: |
| 281 | self._session_prompts.append(user_input) |
| 282 | |
| 283 | console.print() |
| 284 | console.print(" [dim]Thinking...[/]") |
| 285 | |
| 286 | response = await self._agent.run( |
| 287 | prompt=user_input, |
| 288 | on_tool_call=self._handle_tool_call, |
| 289 | ) |
| 290 | |
| 291 | if response.error: |
| 292 | _print_error(response.error) |
| 293 | return |
no outgoing calls
no test coverage detected