emcie-co--parlant/tests/api/test_agents.py

# Copyright 2026 Emcie Co Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from typing import Any
from fastapi import status
import httpx
from lagom import Container
from pytest import mark, raises

from parlant.core.agents import AgentId, AgentStore
from parlant.core.common import ItemNotFoundError
from parlant.core.tags import TagId, TagStore


async def test_that_an_agent_can_be_created_without_description(
    async_client: httpx.AsyncClient,
) -> None:
    response = await async_client.post(
        "/agents",
        json={"name": "test-agent"},
    )

    assert response.status_code == status.HTTP_201_CREATED

    agent = response.json()

    assert agent["name"] == "test-agent"
    assert agent["description"] is None


async def test_that_an_agent_can_be_created_with_description(
    async_client: httpx.AsyncClient,
) -> None:
    response = await async_client.post(
        "/agents",
        json={"name": "test-agent", "description": "You are a test agent"},
    )

    assert response.status_code == status.HTTP_201_CREATED

    agent = response.json()

    assert agent["name"] == "test-agent"
    assert agent["description"] == "You are a test agent"


async def test_that_an_agent_can_be_created_without_max_engine_iterations(
    async_client: httpx.AsyncClient,
) -> None:
    response = await async_client.post(
        "/agents",
        json={"name": "test-agent"},
    )

    assert response.status_code == status.HTTP_201_CREATED

    agent = response.json()

    assert agent["name"] == "test-agent"
    assert agent["max_engine_iterations"] == 3  # Default value


async def test_that_an_agent_can_be_created_with_max_engine_iterations(
    async_client: httpx.AsyncClient,
) -> None:
    response = await async_client.post(
        "/agents",
        json={"name": "test-agent", "max_engine_iterations": 1},
    )

    assert response.status_code == status.HTTP_201_CREATED

    agent = response.json()

    assert agent["name"] == "test-agent"
    assert agent["max_engine_iterations"] == 1


async def test_that_an_agent_can_be_created_with_default_composition_mode(
    async_client: httpx.AsyncClient,
) -> None:
    response = await async_client.post(
        "/agents",
        json={"name": "test-agent"},
    )

    assert response.status_code == status.HTTP_201_CREATED

    agent = response.json()

    assert agent["name"] == "test-agent"
    assert agent["composition_mode"] == "fluid"  # Default mode


async def test_that_an_agent_can_be_created_with_specific_composition_mode(
    async_client: httpx.AsyncClient,
) -> None:
    response = await async_client.post(
        "/agents",
        json={"name": "test-agent", "composition_mode": "strict_canned"},
    )

    assert response.status_code == status.HTTP_201_CREATED

    agent = response.json()

    assert agent["name"] == "test-agent"
    assert agent["composition_mode"] == "strict_canned"


async def test_that_an_agent_can_be_created_with_tags(
    async_client: httpx.AsyncClient,
    container: Container,
) -> None:
    tag_store = container[TagStore]

    tag1 = await tag_store.create_tag("tag1")
    tag2 = await tag_store.create_tag("tag2")

    response = await async_client.post(
        "/agents",
        json={"name": "test-agent", "tags": [tag1.id, tag1.id, tag2.id]},
    )

    assert response.status_code == status.HTTP_201_CREATED

    agent_dto = (
        (await async_client.get(f"/agents/{response.json()['id']}")).raise_for_status().json()
    )

    assert agent_dto["name"] == "test-agent"

    assert len(agent_dto["tags"]) == 2
    assert set(agent_dto["tags"]) == {tag1.id, tag2.id}


async def test_that_an_agent_can_be_listed(
    async_client: httpx.AsyncClient,
) -> None:
    _ = (
        (
            await async_client.post(
                "/agents",
                json={"name": "test-agent"},
            )
        )
        .raise_for_status()
        .json()
    )

    agents = (
        (
            await async_client.get(
                "/agents",
            )
        )
        .raise_for_status()
        .json()
    )

    assert len(agents) == 1
    assert agents[0]["name"] == "test-agent"
    assert agents[0]["description"] is None


async def test_that_an_agent_can_be_read(
    async_client: httpx.AsyncClient,
) -> None:
    agent = (
        (
            await async_client.post(
                "/agents",
                json={"name": "test-agent"},
            )
        )
        .raise_for_status()
        .json()
    )

    agent_dto = (
        (
            await async_client.get(
                f"/agents/{agent['id']}",
            )
        )
        .raise_for_status()
        .json()
    )

    assert agent_dto["name"] == "test-agent"
    assert agent_dto["description"] is None
    assert agent_dto["composition_mode"] == "fluid"


@mark.parametrize(
    "update_payload, expected_name, expected_description, expected_iterations, expected_composition",
    [
        ({"name": "New Name"}, "New Name", None, 3, "fluid"),
        ({"description": None}, "test-agent", None, 3, "fluid"),
        ({"description": "You are a test agent"}, "test-agent", "You are a test agent", 3, "fluid"),
        (
            {"description": "Changed desc", "max_engine_iterations": 2},
            "test-agent",
            "Changed desc",
            2,
            "fluid",
        ),
        ({"max_engine_iterations": 5}, "test-agent", None, 5, "fluid"),
        (
            {"composition_mode": "strict_canned"},
            "test-agent",
            None,
            3,
            "strict_canned",
        ),
    ],
)
async def test_that_an_agent_can_be_updated(
    async_client: httpx.AsyncClient,
    container: Container,
    update_payload: dict[str, Any],
    expected_name: str,
    expected_description: str | None,
    expected_iterations: int,
    expected_composition: str,
) -> None:
    agent_store = container[AgentStore]
    agent = await agent_store.create_agent("test-agent")

    response = await async_client.patch(f"/agents/{agent.id}", json=update_payload)
    response.raise_for_status()
    updated_agent = response.json()

    assert updated_agent["name"] == update_payload.get("name", "test-agent")
    assert updated_agent["name"] == expected_name
    assert updated_agent["description"] == expected_description
    assert updated_agent["max_engine_iterations"] == expected_iterations
    assert updated_agent["composition_mode"] == expected_composition


async def test_that_an_agent_can_be_deleted(
    async_client: httpx.AsyncClient,
    container: Container,
) -> None:
    agent_store = container[AgentStore]
    agent = await agent_store.create_agent("test-agent")

    delete_response = await async_client.delete(
        f"/agents/{agent.id}",
    )
    assert delete_response.status_code == status.HTTP_204_NO_CONTENT

    with raises(ItemNotFoundError):
        await agent_store.read_agent(agent.id)


async def test_that_tags_can_be_added_to_an_agent(
    async_client: httpx.AsyncClient,
    container: Container,
) -> None:
    agent_store = container[AgentStore]
    tag_store = container[TagStore]

    tag1 = await tag_store.create_tag("tag1")
    tag2 = await tag_store.create_tag("tag2")

    agent = await agent_store.create_agent("test-agent")

    update_payload = {"tags": {"add": [tag1.id, tag2.id]}}
    response = await async_client.patch(f"/agents/{agent.id}", json=update_payload)
    response.raise_for_status()
    updated_agent = response.json()

    assert tag1.id in updated_agent["tags"]
    assert tag2.id in updated_agent["tags"]

    agent_dto = (await async_client.get(f"/agents/{agent.id}")).raise_for_status().json()
    assert tag1.id in agent_dto["tags"]
    assert tag2.id in agent_dto["tags"]


async def test_that_tags_can_be_removed_from_an_agent(
    async_client: httpx.AsyncClient,
    container: Container,
) -> None:
    agent_store = container[AgentStore]
    agent = await agent_store.create_agent("test-agent")

    await agent_store.upsert_tag(agent.id, TagId("tag1"))
    await agent_store.upsert_tag(agent.id, TagId("tag2"))
    await agent_store.upsert_tag(agent.id, TagId("tag3"))

    update_payload = {"tags": {"remove": ["tag1", "tag3"]}}
    response = await async_client.patch(f"/agents/{agent.id}", json=update_payload)
    response.raise_for_status()
    updated_agent = response.json()

    assert "tag1" not in updated_agent["tags"]
    assert "tag2" in updated_agent["tags"]
    assert "tag3" not in updated_agent["tags"]


async def test_that_tags_can_be_added_and_removed_in_same_request(
    async_client: httpx.AsyncClient,
    container: Container,
) -> None:
    agent_store = container[AgentStore]
    tag_store = container[TagStore]

    tag1 = await tag_store.create_tag("tag1")
    tag2 = await tag_store.create_tag("tag2")
    tag3 = await tag_store.create_tag("tag3")
    tag4 = await tag_store.create_tag("tag4")

    agent = await agent_store.create_agent("test-agent")

    await agent_store.upsert_tag(agent.id, tag1.id)
    await agent_store.upsert_tag(agent.id, tag2.id)

    update_payload = {"tags": {"add": [tag3.id, tag4.id], "remove": [tag1.id]}}
    response = await async_client.patch(f"/agents/{agent.id}", json=update_payload)
    response.raise_for_status()
    updated_agent = response.json()

    assert tag1.id not in updated_agent["tags"]
    assert tag2.id in updated_agent["tags"]
    assert tag3.id in updated_agent["tags"]
    assert tag4.id in updated_agent["tags"]


async def test_that_an_agent_cannot_be_created_with_a_nonexistent_tag(
    async_client: httpx.AsyncClient,
    container: Container,
) -> None:
    agent_store = container[AgentStore]

    agent = await agent_store.create_agent("test-agent")

    response = await async_client.patch(
        f"/agents/{agent.id}",
        json={"tags": {"add": ["nonexistent-tag"]}},
    )

    assert response.status_code == status.HTTP_404_NOT_FOUND


async def test_that_an_agent_can_be_created_with_custom_id(
    async_client: httpx.AsyncClient,
    container: Container,
) -> None:
    custom_id = "my_custom_agent_id"
    name = "Custom ID Agent"
    description = "An agent with a custom ID"

    response = await async_client.post(
        "/agents",
        json={
            "name": name,
            "description": description,
            "id": custom_id,
        },
    )

    assert response.status_code == status.HTTP_201_CREATED

    agent = response.json()
    assert agent["id"] == custom_id
    assert agent["name"] == name
    assert agent["description"] == description


async def test_that_multiple_agents_can_be_created_with_different_custom_ids(
    async_client: httpx.AsyncClient,
) -> None:
    # Create first agent with custom ID
    response1 = await async_client.post(
        "/agents",
        json={
            "name": "First Agent",
            "description": "First agent",
            "id": "agent_1",
        },
    )
    assert response1.status_code == status.HTTP_201_CREATED
    assert response1.json()["id"] == "agent_1"

    # Create second agent with different custom ID
    response2 = await async_client.post(
        "/agents",
        json={
            "name": "Second Agent",
            "description": "Second agent",
            "id": "agent_2",
        },
    )
    assert response2.status_code == status.HTTP_201_CREATED
    assert response2.json()["id"] == "agent_2"


async def test_that_creating_agent_with_duplicate_custom_id_fails(
    async_client: httpx.AsyncClient,
    container: Container,
) -> None:
    custom_id = AgentId("duplicate_agent_id")

    # Create first agent with custom ID
    response1 = await async_client.post(
        "/agents",
        json={
            "name": "First Agent",
            "description": "First agent",
            "id": custom_id,
        },
    )
    assert response1.status_code == status.HTTP_201_CREATED
    assert response1.json()["id"] == custom_id

    # Try to create second agent with same ID at the store level - should fail
    agent_store = container[AgentStore]
    with raises(ValueError, match="already exists"):
        await agent_store.create_agent(
            name="Second Agent",
            description="Second agent",
            id=custom_id,
        )


async def test_that_agent_composition_mode_can_be_set_and_updated(
    async_client: httpx.AsyncClient,
) -> None:
    # Create agent with CANNED_COMPOSITED mode
    response = await async_client.post(
        "/agents",
        json={
            "name": "test-agent",
            "composition_mode": "composited_canned",
        },
    )

    assert response.status_code == status.HTTP_201_CREATED
    agent = response.json()
    agent_id = agent["id"]

    # Check that the composition mode is set correctly after creation
    assert agent["composition_mode"] == "composited_canned"

    # Retrieve agent and verify composition mode
    response = await async_client.get(f"/agents/{agent_id}")
    assert response.status_code == status.HTTP_200_OK
    agent = response.json()
    assert agent["composition_mode"] == "composited_canned"

    # Update agent to CANNED_STRICT mode
    response = await async_client.patch(
        f"/agents/{agent_id}",
        json={
            "composition_mode": "strict_canned",
        },
    )

    assert response.status_code == status.HTTP_200_OK
    agent = response.json()

    # Check that the composition mode is updated correctly
    assert agent["composition_mode"] == "strict_canned"

    # Retrieve agent again and verify composition mode
    response = await async_client.get(f"/agents/{agent_id}")
    assert response.status_code == status.HTTP_200_OK
    agent = response.json()
    assert agent["composition_mode"] == "strict_canned"