(
contents: list[str] = Field(description="List of observations or memories to store"),
)
| 264 | |
| 265 | @mcp.tool() |
| 266 | async def remember( |
| 267 | contents: list[str] = Field(description="List of observations or memories to store"), |
| 268 | ): |
| 269 | deps = Deps(openai=AsyncOpenAI(), pool=await get_db_pool()) |
| 270 | try: |
| 271 | return "\n".join(await asyncio.gather(*[add_memory(content, deps) for content in contents])) |
| 272 | finally: |
| 273 | await deps.pool.close() |
| 274 | |
| 275 | |
| 276 | @mcp.tool() |
nothing calls this directly
no test coverage detected